Calculate distance flown using Python

Let's say we want to be able to calculate the distance that an aircraft flew, and over which countries it flew.

In this example, we will go through the steps necessary to recover the data from a flight and calculate the distance flown during the flight using the Tracking History API.

Source code#

You can find and download the source code for this example here.

Geo filtering#

First, we will look at a specific airport, from which we will single out a specific flight to analyze.

Let's take CDG airport here. We will look at this GeoJSON representing the area of the airport:

{
  "type": "FeatureCollection",
  "features": [
    {
      "type": "Feature",
      "properties": {},
      "geometry": {
        "type": "Polygon",
        "coordinates": [
          [
            [
              2.481443,
              48.970752
            ],
            [
              2.642431,
              48.970752
            ],
            [
              2.642431,
              49.041694
            ],
            [
              2.481443,
              49.041694
            ],
            [
              2.481443,
              48.970752
            ]
          ]
        ]
      }
    }
  ]
}

Which is represented like that on a map:

Calling the Tracking History API and pick a flight#

Let's start by installing the request, pandas and colorama packages that will be necessary for this first step to work.

pip install requests pandas colorama

Using the coordinates previously shown, we can filter the API results for the area we want:

Latitude range: 48.970752,49.041694 need to be South to North, so values can vary from -90 to 90.
Longitude range: 2.481443,2.642431 need to be West to East, so values can vary from -170 to 170.

import os
import yaml
import requests
import json
import pandas as pd
from colorama import init, Fore, Style, Back


if __name__ == "__main__":
    # Initialize colorama package
    init()

    # Finding target updates around CDG airport
    print(Fore.YELLOW + f"Loading the datapoints at the airport ...")
    resp = requests.get(
        "https://api.airsafe.spire.com/v2/targets/history",
        params={
            "start": "2021-04-20T12:00:00Z",
            "end": "2021-04-20T18:30:00Z",
            # Coordinates around CDG airport
            "longitude_between": "2.481443,2.642431",
            "latitude_between": "48.970752,49.041694",
        },
        headers={"Authorization": f"Bearer <your_token>"},
    )

    if resp.status_code == 401:
        print(Style.RESET_ALL + Fore.RED + "invalid token.")
        exit()
    else:
        data = []
        for line in resp.iter_lines(decode_unicode=True):
            if line and '"target":{' in line:
                data.append(json.loads(line)["target"])
        df = pd.DataFrame(data)

    # Find a target update that fits the route we are interested in
    flight_analysed = df[
        (df.arrival_airport_iata == "CDG") & (df.departure_airport_iata == "LHR")
    ]

Let's breakdown what we did here:

We queried our endpoint around CDG airport during a specific time.
We parsed the results and generated a Pandas Dataframe from them.
We filtered out all flights with the route LHR to CDG.

Get the picked flight data and calculate distance flown#

From the filtered flights fitting our requirement (LHR to CDG), we pick out the first of those and query the Tracking History endpoint again, this time removing geo limitations. For this work out, we will need to install an additional package: geopy

pip install geopy

import os
import yaml
import requests
import json
import pandas as pd
from geopy import distance
from colorama import init, Fore, Style, Back


if __name__ == "__main__":
    init()

    # Finding target updates around CDG airport
    print(Fore.YELLOW + f"Loading the datapoints at the airport ...")
    resp = requests.get(
        "https://api.airsafe.spire.com/v2/targets/history",
        params={
            "start": "2021-04-20T12:00:00Z",
            "end": "2021-04-20T18:30:00Z",
            # Coordinates around CDG airport
            "longitude_between": "2.481443,2.642431",
            "latitude_between": "48.970752,49.041694",
        },
        headers={"Authorization": f"Bearer <your_token>"},
    )

    if resp.status_code == 401:
        print(Style.RESET_ALL + Fore.RED + "invalid token.")
        exit()
    else:
        data = []
        for line in resp.iter_lines(decode_unicode=True):
            if line and '"target":{' in line:
                data.append(json.loads(line)["target"])
        df = pd.DataFrame(data)

    # Find a target update that fits the route we are interested in
    flight_analysed = df[
        (df.arrival_airport_iata == "CDG") & (df.departure_airport_iata == "LHR")
    ]

    # If we have a flight fitting the destination/arrival we want
    if not flight_analysed.empty:
        print(Fore.YELLOW + f"A flight has been found, loading flight path ...")
        print(
            Style.RESET_ALL + Fore.GREEN + f"Flight number:" + Style.RESET_ALL,
            Back.RED + f"{flight_analysed.iloc[0]['flight_number']}",
        )
        print(
            Style.RESET_ALL + Fore.GREEN + f"Flight from/to:" + Style.RESET_ALL,
            Back.RED
            + f"{flight_analysed.iloc[0]['departure_airport_iata']}/{flight_analysed.iloc[0]['arrival_airport_iata']}",
        )
        print(
            Style.RESET_ALL + Fore.GREEN + f"Flight scheduled time:" + Style.RESET_ALL,
            Back.RED
            + f"{flight_analysed.iloc[0]['departure_scheduled_time']} - {flight_analysed.iloc[0]['arrival_scheduled_time']}",
        )
        # Requesting flight path for the ICAO address during the specified flight period
        resp = requests.get(
            "https://api.airsafe.spire.com/v2/targets/history",
            params={
                "start": flight_analysed.iloc[0]["departure_scheduled_time"],
                "end": flight_analysed.iloc[0]["arrival_scheduled_time"],
                # Tracking the icao address of this aircraft only
                "icao_address": flight_analysed.iloc[0]["icao_address"],
            },
            headers={"Authorization": f"Bearer <your_token>"},
        )

        data = []
        for line in resp.iter_lines(decode_unicode=True):
            if line and '"target":{' in line:
                data.append(json.loads(line)["target"])
        df = pd.DataFrame(data)

        print(
            Style.RESET_ALL + Fore.GREEN + f"Datapoints found:" + Style.RESET_ALL,
            Back.RED + f"{len(df.index)}",
        )
        print(Style.RESET_ALL)
        print(
            f"Earliest point found at: {df['timestamp'].head(1).to_string(index=False)}"
        )
        print(
            f"Latest point found at: {df['timestamp'].tail(1).to_string(index=False)}"
        )

        total_distance_km = 0

        row_iterator = df.iterrows()
        _, last = next(row_iterator)
        for i, row in row_iterator:
            coord_last = (last["latitude"], last["longitude"])
            coord_current = (row["latitude"], row["longitude"])
            last = row
            # Using geopy and the geodesic distance between 2 points, we can calculate the total distance
            total_distance_km += distance.distance(coord_last, coord_current).km

        print(Style.RESET_ALL)
        print(
            Fore.GREEN + f"\bTotal distance flown:" + Style.RESET_ALL,
            Back.RED + f"{total_distance_km} km",
        )
    else:
        print(Style.RESET_ALL + Fore.RED + "Have not found the flight path.")
        exit()

Let's break down what we did here:

We selected the first flight in our filtered data flight_analysed.iloc[0], and queried the Tracking History endpoint to get this specific flight using its ICAO address as the API filter.
We generated a Panda Dataframe with the retrieved data.
We iterate over every data point, and generated an array of coordinates, which we will use to add up the distance between every points. For that we will use the geopy package, that calculates distance between 2 sets of coordinates.

Get countries flown over by the flight#

We can now make use of the reverse_geocoder package, which is an offline geocoder that will allow us to get useful information from coordinates such as the country.

pip install reverse_geocoder

import os
import yaml
import requests
import json
import pandas as pd
from geopy import distance
from colorama import init, Fore, Style, Back
import reverse_geocoder as rg


if __name__ == "__main__":
    init()

    # Finding target updates around CDG airport
    print(Fore.YELLOW + f"Loading the datapoints at the airport ...")
    resp = requests.get(
        "https://api.airsafe.spire.com/v2/targets/history",
        params={
            "start": "2021-04-20T12:00:00Z",
            "end": "2021-04-20T18:30:00Z",
            # Coordinates around CDG airport
            "longitude_between": "2.481443,2.642431",
            "latitude_between": "48.970752,49.041694",
        },
        headers={"Authorization": f"Bearer <your_token>"},
    )

    if resp.status_code == 401:
        print(Style.RESET_ALL + Fore.RED + "invalid token.")
        exit()
    else:
        data = []
        for line in resp.iter_lines(decode_unicode=True):
            if line and '"target":{' in line:
                data.append(json.loads(line)["target"])
        df = pd.DataFrame(data)

    # Find a target update that fits the route we are interested in
    flight_analysed = df[
        (df.arrival_airport_iata == "CDG") & (df.departure_airport_iata == "LHR")
    ]

    # If we have a flight fitting the destination/arrival we want
    if not flight_analysed.empty:
        print(Fore.YELLOW + f"A flight has been found, loading flight path ...")
        print(
            Style.RESET_ALL + Fore.GREEN + f"Flight number:" + Style.RESET_ALL,
            Back.RED + f"{flight_analysed.iloc[0]['flight_number']}",
        )
        print(
            Style.RESET_ALL + Fore.GREEN + f"Flight from/to:" + Style.RESET_ALL,
            Back.RED
            + f"{flight_analysed.iloc[0]['departure_airport_iata']}/{flight_analysed.iloc[0]['arrival_airport_iata']}",
        )
        print(
            Style.RESET_ALL + Fore.GREEN + f"Flight scheduled time:" + Style.RESET_ALL,
            Back.RED
            + f"{flight_analysed.iloc[0]['departure_scheduled_time']} - {flight_analysed.iloc[0]['arrival_scheduled_time']}",
        )
        # Requesting flight path for the ICAO address during the specified flight period
        resp = requests.get(
            "https://api.airsafe.spire.com/v2/targets/history",
            params={
                "start": flight_analysed.iloc[0]["departure_scheduled_time"],
                "end": flight_analysed.iloc[0]["arrival_scheduled_time"],
                # Tracking the icao address of this aircraft only
                "icao_address": flight_analysed.iloc[0]["icao_address"],
            },
            headers={"Authorization": f"Bearer <your_token>"},
        )

        data = []
        for line in resp.iter_lines(decode_unicode=True):
            if line and '"target":{' in line:
                data.append(json.loads(line)["target"])
        df = pd.DataFrame(data)

        print(
            Style.RESET_ALL + Fore.GREEN + f"Datapoints found:" + Style.RESET_ALL,
            Back.RED + f"{len(df.index)}",
        )
        print(Style.RESET_ALL)
        print(
            f"Earliest point found at: {df['timestamp'].head(1).to_string(index=False)}"
        )
        print(
            f"Latest point found at: {df['timestamp'].tail(1).to_string(index=False)}"
        )

        total_distance_km = 0

        row_iterator = df.iterrows()
        _, last = next(row_iterator)
        coordinates = []
        for i, row in row_iterator:
            coord_last = (last["latitude"], last["longitude"])
            coordinates.append(coord_last)
            coord_current = (row["latitude"], row["longitude"])
            last = row
            # Using geopy and the geodesic distance between 2 points, we can calculate the total distance
            total_distance_km += distance.distance(coord_last, coord_current).km
        coordinates.append(coord_current)

        print(Style.RESET_ALL)
        # We can now check the countries flown by by using a reverse offline geocoder
        results = rg.search(coordinates)
        df = pd.DataFrame(results)
        print(Fore.GREEN + f"Countries flown:" + Style.RESET_ALL, df["cc"].unique())

        print(Style.RESET_ALL)
        print(
            Fore.GREEN + f"\bTotal distance flown:" + Style.RESET_ALL,
            Back.RED + f"{total_distance_km} km",
        )
    else:
        print(Style.RESET_ALL + Fore.RED + "Have not found the flight path.")
        exit()

Output#

Here is the expected output from our example:

Loading the datapoints at the airport ...
A flight has been found, loading flight path ...
Flight number: AF1281
Flight from/to: LHR/CDG
Flight scheduled time: 2021-04-20T16:35:00Z - 2021-04-20T17:55:00Z
Datapoints found: 653

Earliest point found at: 2021-04-20T16:35:00Z
Latest point found at: 2021-04-20T17:24:16Z

Loading formatted geocoded file...
Countries flown: ['GB' 'FR']

Total distance flown: 383.0209466161545 km