traffic.algorithms.navigation

Alignment detection

class traffic.algorithms.navigation.alignment.BeaconTrackBearingAlignment(points, angle_precision=1, time_precision='2 min', min_time='30s', min_distance=80)

Iterates on segments of trajectories aligned with one of the given navigational beacons passed in parameter.

The name of the navigational beacon is assigned in a new column navaid.

Parameters:

points (Union[str, PointMixin, Iterable[PointMixin], FlightPlan]) – a set of point or identifiers constituting the flight plan. A FlightPlan structure can also be provided.
angle_precision (int) – in degrees. The difference between the track angle and the bearing with respect to the point must be less than this threshold.
time_precision (str) – The maximum time interval during which the difference between the angles can exceed the threshold.
min_time (str) – Only segments with a duration of at least min_time will be yielded.
min_distance (int) – The minimal distance to a given point must be above the min_distance value to be returned.

Usage:

>>> from traffic.data import navaids
>>> from traffic.data.samples import elal747
>>> subset = elal747.skip("2h30min").first("2h30min")
>>> for segment in subset.aligned([navaids['KAVOS'], navaids['PEDER']]):
...     print(f"aligned on {segment.navaid_max} for {segment.duration}")
aligned on KAVOS for 0 days 00:07:00
aligned on PEDER for 0 days 00:05:40

Go-around detection

class traffic.algorithms.navigation.go_around.GoAroundDetection(airport=None, dataset=None)

Detects go-arounds.

The method yields pieces of trajectories with exactly two landing attempts (aligned on one runway) on the same airport separated by exactly one climbing phase.

Parameters:

airport (None | str | Airport) – If None, the method tries to guess the landing airport based on the dataset parameter. (see landing_airport())
dataset (Optional[Airports]) – database of candidate airports, only used if airport is None

Example usage:

>>> from traffic.data.samples import belevingsvlucht

By default, go arounds will be detected at Amsterdam Schiphol airport (EHAM)

>>> belevingsvlucht.go_around().next()
>>> amsterdam_goaround = GoAroundDetection(airport="EHAM")
>>> belevingsvlucht.go_around(method=amsterdam_goaround).next()

There were none; however we find 5 of them at Lelystad airport.

>>> belevingsvlucht.go_around("EHLE").sum()
5

Landing detection

class traffic.algorithms.navigation.landing.LandingAlignedOnILS(airport, angle_tolerance=0.1, min_duration='1 min', max_ft_above_airport=5000)

Iterates on all segments of trajectory aligned with the ILS of the given airport. The runway number is appended as a new ILS column.

Parameters:

airport (str | Airport) – Airport where the ILS is located
angle_tolerance (float) – maximum tolerance on bearing difference between ILS and flight trajectory.
min_duration (Union[None, str, Real, timedelta, Timedelta]) – minimum duration a flight has to spend on the ILS to be considered as aligned.
max_ft_above_airport (float) – maximum altitude AGL, relative to the airport, that a flight can be to be considered as aligned.

Example usage:

>>> from traffic.data.samples import belevingsvlucht

The detected runway identifier is available in the ILS column:

>>> aligned = belevingsvlucht.landing('EHAM').next()
>>> f"ILS {aligned.max('ILS')}; landing time {aligned.stop:%H:%M}"
'ILS 06; landing time 20:17'

We can specify the method by default:

>>> for aligned in belevingsvlucht.landing('EHLE', method="default"):
...     print(aligned.start)
2018-05-30 16:00:55+00:00
2018-05-30 16:50:44+00:00
2018-05-30 17:21:17+00:00
2018-05-30 18:13:02+00:00
2018-05-30 19:05:22+00:00
2018-05-30 19:42:36+00:00

>>> final = belevingsvlucht.final("landing('EHLE', method='default')")
>>> final.ILS_max  # equivalent to final.max("ILS")
'23'

Usual built-in functions works as on any kind of iterators, here we get the flight segment with the latest start timestamp. (equivalent to the final method used above)

>>> from operator import attrgetter
>>> last_aligned = max(
...     belevingsvlucht.landing("EHLE"),
...     key=attrgetter('start')
... )
>>> last_aligned.start
Timestamp('2018-05-30 19:42:36+0000', tz='UTC')

class traffic.algorithms.navigation.landing.LandingAnyAttempt(dataset=None, alt_threshold=8000, **kwargs)

Iterates on landing attempts without specific airport information.

First, candidates airports are identified in the neighbourhood of the segments of trajectory below a threshold altitude. By default, the full airport database is considered but it is possible to restrict it and pass a smaller database with the dataset parameter.

If no runway information is available for the given airport, no trajectory segment will be provided.

Parameters:

dataset (Optional[Airports]) – a subset of airports used as candidate landing airports
alt_threshold (float) – an altitude threshold helpful to select candidate airports. It is important to pick a threshold significantly above the altitude of the candidate airports.

>>> from traffic.data.samples import belevingsvlucht
>>> attempts = belevingsvlucht.landing(method="any")
>>> for i, attempt in enumerate(attempts):
...     print(f"Step {i}: {attempt.airport_max} runway {attempt.ILS_max}")
Step 0: EHLE runway 23
Step 1: EHLE runway 05
Step 2: EHLE runway 23
Step 3: EHLE runway 05
Step 4: EHLE runway 23
Step 5: EHAM runway 06

class traffic.algorithms.navigation.landing.LandingWithRunwayChange(airport, **kwargs)

Detects runway changes.

The method yields pieces of trajectories with exactly two runway alignments on the same airport not separated by a climbing phase.

In each piece of yielded trajectory, the ILS column contains the name of the runway targetted by the aircraft at each instant.

Parameters:

airport (str | Airport) – Airport where the ILS is located
kwargs (Any) – are passed to the constructor of LandingAlignedOnILS

We can reuse the trajectory from the Getting started page to illustrate the usage of this method

>>> from traffic.data.samples import quickstart
>>> flight = quickstart['AFR17YC']
>>> flight.has("landing(airport='LFPG', method='runway_change')")
True

Since the trajectory has a runway change, we can print characteristics of each segment:

>>> segments = flight.landing('LFPG', method="aligned_on_ils")
>>> first = next(segments)
>>> f"{first.start:%H:%M} {first.stop:%H:%M} aligned on {first.ILS_max}"
'13:34 13:36 aligned on 08R'
>>> second = next(segments)
>>> f"{second.start:%H:%M} {second.stop:%H:%M} aligned on {second.ILS_max}"
'13:36 13:39 aligned on 08L'

Flight phase detection

class traffic.algorithms.navigation.phases.FlightPhasesOpenAP(twindow=60)

Assign a flight phase to each timestamp of a flight

This implementation uses the OpenAP phase detection fuzzy logic method.

Usage:

>>> from traffic.data.samples import belevingsvlucht
>>> climb = belevingsvlucht.phases().query('phase == "CLIMB"')
>>> first_climb = climb.next("split")

We can confirm the non-empty intersection between the climb and take-off phases. The intersection operator & returns the part of the trajectory which is simultaneously a take-off and a climb:

>>> takeoff = belevingsvlucht.next('takeoff("EHAM")')
>>> (takeoff & first_climb).duration
Timedelta('0 days 00:00:23')

Holding pattern detection

class traffic.algorithms.navigation.holding_pattern.MLHoldingDetection(duration='6 min', step='2 min', threshold='5 min', samples=30, model_path=None, vertical_rate=False)

Iterates on all holding pattern segments in the trajectory.

This approach is based on a neuronal network model. Details will be published in a coming academic publication.

Parameters should be left as default as they are strongly coupled with the proposed model.

The model has been trained on manually labelled holding patterns for trajectories landing at different European airports including London Heathrow.

All the parameters below are closely intricated with a model. Default parameters should not be changed if we stick with the provided model.

Parameters:

duration (str) – the duration of each sliding window
step (str) – the step for each sliding window
threshold (str) – the minimum duration for each sliding window
samples (int) – each sliding window is resampled to a fixed number of points
model_path (None | str | Path) – where the models are located, by default in the same module directory
vertical_rate (bool) – a boolean set to True if the vertical rate should be taken into account in the model.

>>> from traffic.data.samples import belevingsvlucht
>>> belevingsvlucht.has("holding_pattern")
True

(new in version 2.8)

Point-merge detection

class traffic.algorithms.navigation.point_merge.PointMerge(point_merge, secondary_point=None, distance_interval=None, delta_threshold=0.05, airport=None, runway=None, **kwargs)

Iterates on all point merge segments in a trajectory before landing at a given airport.

Only the point_merge argument is mandatory but other arguments may reduce the number of false positives.

Parameters:

point_merge (str | PointMixin | list[PointMergeParams]) – The procedure point on which trajectories all align.
secondary_point (None | str | PointMixin) –
In some cases (e.g. Dublin 10R), aircraft align to the point_merge after a segment of almost constant distance to a secondary point.

Most often, the secondary_point is the point_merge and can be left as None.
distance_interval (None | tuple[float, float]) –
A tuple of distances in nautical miles, corresponding to lower and upper bound distances in the AIP between the constant distance segments and the point merge.

This parameter is ignored if left as None.
delta_threshold (float) – keep as default
airport (None | str | Airport) – Remove false positives by specifying the landing airport. The algorithm will ensure all trajectories are aligned with one of the airport’s ILS.
runway (None | str) – Remove false positives by specifying the landing runway. The algorithm will ensure all trajectories are aligned with the runway’s ILS. (ignored if airport is None)

Usage: See How to implement point-merge detection?

(new in version 2.8)

class traffic.algorithms.navigation.point_merge.PointMergeParams

Take-off detection

class traffic.algorithms.navigation.takeoff.PolygonBasedRunwayDetection(airport, max_ft_above_airport=2000, zone_length=6000, little_base=50, opening=5)

Identifies the take-off runway for trajectories.

Iterates on all segments of trajectory matching a zone around a runway of the given airport. The takeoff runway number is appended as a new runway column.

Parameters:

airport (str | Airport) – The airport from where the flight takes off.
max_ft_above_airport (float) – maximum altitude AGL, relative to the airport, that a flight can be to be considered as aligned.
zone_length (int) – the length of the trapeze, aligned with the runway.
little_base (int) – the smallest base of the trapeze, on the runway threshold side.
opening (float) – the angle (in degrees) of opening of the trapeze.

>>> from traffic.data.samples import belevingsvlucht
>>> takeoff = belevingsvlucht.next('takeoff("EHAM", method="default")')
>>> takeoff.duration
Timedelta('0 days 00:00:25')

class traffic.algorithms.navigation.takeoff.TrackBasedRunwayDetection(airport, max_ft_above_airport=1500, min_groundspeed_kts=30, min_vert_rate_ftmin=257, max_bearing_deg=10, max_dist_nm=5)

Determines the taking-off runway of a flight based on its surface trajectory

Parameters:

airport (str | Airport) – The airport from where the flight takes off.
max_ft_above_airport (float) – maximum altitude AGL, relative to the airport, that a flight can be to be considered as aligned.
min_groundspeed_kts (float) – Minimum groundspeed to consider.
min_vert_rate_ftmin (float) – Minimum vertical rate to consider.
maximum_bearing_deg – Maximum bearing difference to consider.
max_dist_nm (float) – Maximum distance from the airport to consider.

>>> from traffic.data.samples import elal747
>>> takeoff = elal747.takeoff(method="track_based", airport="LIRF").next()
>>> takeoff.duration, takeoff.runway_max
(Timedelta('0 days 00:00:40'), '25')

>>> from traffic.data.samples import belevingsvlucht
>>> takeoff = belevingsvlucht.next('takeoff("EHAM", method="default")')
>>> takeoff.duration
Timedelta('0 days 00:00:25')

Protocols

class traffic.algorithms.navigation.ApplyBase(*args, **kwargs): Classes following this protocol should implement an apply method which returns a Flight.

class traffic.algorithms.navigation.ApplyIteratorBase(*args, **kwargs): Classes following this protocol should implement an apply method which returns an iterator of Flight.

class traffic.algorithms.navigation.ApplyOptionalBase(*args, **kwargs): Classes following this protocol should implement an apply method which returns None or a Flight.

traffic.algorithms.navigation

Alignment detection

Go-around detection

Landing detection

Flight phase detection

Holding pattern detection

Point-merge detection

Take-off detection

Bonus: Thermals detection

Protocols