WIP Flights Overhead Trajectory Intersection Calculator

FlightsOverhead.py

#!/usr/bin/env python3
from FlightRadar24 import FlightRadar24API
from flydenity import Parser as FlydenityParser
import airportsdata
from bs4 import BeautifulSoup
import requests
from pprint import pprint
import time
import math
import folium
from folium.plugins import PolyLineTextPath

NEWARK = [ 40.689306 , -74.173820 ]
NM_TO_KM = 1.852 # Conversion factor from knots to kilometers per hour
EARTH_RADIUS = 6371  # Earth radius in kilometers

def haversine( lat1 , lon1 , lat2 , lon2 ):
	lat1, lon1, lat2, lon2 = map(math.radians, [lat1, lon1, lat2, lon2])
	dlat = lat2 - lat1
	dlon = lon2 - lon1
	a = math.sin(dlat/2)**2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon/2)**2
	c = 2 * math.atan2(math.sqrt(a), math.sqrt(1-a))
	result = ( EARTH_RADIUS * c )
	return result

def calculate_intersection(center, radius, lat, lon, heading, ground_speed, vertical_speed):
	# Calculate the great-circle distance to the center point
	distance_to_center = haversine(lat, lon, center[0], center[1])

	# Convert heading to radians
	heading_rad = math.radians(heading)

	# Calculate the bearing from the current position to the center point
	bearing_to_center = math.atan2(
		math.sin(math.radians(center[1] - lon)) * math.cos(math.radians(center[0])),
		math.cos(math.radians(lat)) * math.sin(math.radians(center[0])) -
		math.sin(math.radians(lat)) * math.cos(math.radians(center[0])) * math.cos(math.radians(center[1] - lon))
	)

	# Normalize bearing_to_center
	bearing_to_center = (bearing_to_center + 2 * math.pi) % (2 * math.pi)

	# Calculate the angular difference between the flight's heading and the bearing to the center point
	angle_difference = abs(heading_rad - bearing_to_center)

	# Ensure the angle difference is within [-π, π]
	if angle_difference > math.pi:
		angle_difference = 2 * math.pi - angle_difference

	# Calculate the closest approach distance
	closest_approach = distance_to_center * math.sin(angle_difference)

	if closest_approach <= radius:
		# Check if the projection is forward
		forward_projection = math.cos(angle_difference) > 0
		if forward_projection:
			# Calculate time to arrive based on ground speed (horizontal speed)
			horizontal_speed_kmh = ground_speed * NM_TO_KM
			time_to_arrive_seconds = (distance_to_center / horizontal_speed_kmh) * 3600  # Convert hours to seconds

			# Calculate vertical distance and time based on vertical speed
			vertical_speed_ms = vertical_speed * 0.00508  # Convert ft/min to m/s
			vertical_distance = time_to_arrive_seconds * vertical_speed_ms  # Time in seconds * vertical speed in m/s

			return True, distance_to_center, time_to_arrive_seconds, vertical_distance, heading_rad
	return False, None, None, None, None

def plot_flight_trajectories(center, radius, flights):
	# Create a map centered at the center point
	m = folium.Map(location=center, zoom_start=6)

	# Add the center point to the map
	folium.Marker(center, popup='Center Point', icon=folium.Icon(color='blue')).add_to(m)

	# Add the circle with the specified radius
	folium.Circle(center, radius=radius*1000, color='blue', fill=False).add_to(m)

	for flight in flights:
		lat = flight["trajectory"]["lat"]
		lon = flight["trajectory"]["lng"]
		heading = flight["trajectory"]["heading"]
		intersects = flight["trajectory"]["intersects"]
		time_to_arrive = flight["trajectory"]["time_to_arrive"]
		flight_number = flight["aircraft"]["tail_number"]
		registration = flight["aircraft"]["registration"]

		# Calculate the end point of the trajectory
		distance = 10000  # Arbitrary large distance for plotting the trajectory line
		end_lat = lat + (distance / EARTH_RADIUS) * (180 / math.pi) * math.cos(math.radians(heading))
		end_lon = lon + (distance / EARTH_RADIUS) * (180 / math.pi) * math.sin(math.radians(heading)) / math.cos(math.radians(lat))

		# Determine the color of the trajectory line
		color = 'red' if intersects else 'green'

		# Add the trajectory line to the map
		polyline = folium.PolyLine([(lat, lon), (end_lat, end_lon)], color=color, weight=2.5, opacity=1).add_to(m)

		# Add an arrow to indicate the direction
		PolyLineTextPath(
			polyline,
			' ➤ ',
			repeat=True,
			offset=12,
			attributes={'fill': color, 'font-weight': 'bold', 'font-size': '16'}
		).add_to(m)

		if intersects:
			label_text = f"{flight_number} / {registration} : {time_to_arrive}"
			folium.Marker(
				location=(lat, lon),
				icon=folium.DivIcon(
					icon_size=(150,36),
					icon_anchor=(0,0),
					html=f'<div style="font-size: 12pt; color: {color};">{label_text}</div>',
				)
			).add_to(m)

	return m

def parse_aircraft_info( html_content ):
	soup = BeautifulSoup(html_content, 'html.parser')
	aircraft_info = {}
	# Find all tables that have a class indicating they are data tables
	tables = soup.find_all('table', class_='devkit-table')
	for table in tables:
		# Use the caption as the key for each section of data
		caption = table.find('caption', class_='devkit-table-title')
		if caption:
			title = caption.text.strip()
			aircraft_info[title] = {}
			rows = table.find_all('tr', class_='devkit-table-row')
			for row in rows:
				cells = row.find_all('td')
				# Handle rows based on the number of cells they contain
				if len(cells) > 1:  # Ignore rows with only one cell unless it's needed
					for cell in cells:
						label = cell.get('data-label', '').strip()  # Handle missing 'data-label'
						if label:  # Only store data that has a label
							aircraft_info[title][label] = cell.text.strip()
				elif len(cells) == 1:  # Handle single cell rows if needed
					info = cells[0].text.strip()
					if info == "None":
						continue
					if info.startswith( "The information contained" ):
						continue
					aircraft_info[title]['General Information'] = info

	return aircraft_info

def search_n_number( n_number ):
	if n_number.startswith( "N" ):
		n_number = n_number[ 1: ]
	url = 'https://registry.faa.gov/aircraftinquiry/Search/NNumberResult'
	headers = {
		'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:127.0) Gecko/20100101 Firefox/127.0',
		'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8',
		'Accept-Language': 'en-US,en;q=0.5',
		'Accept-Encoding': 'gzip, deflate, br, zstd',
		'Content-Type': 'application/x-www-form-urlencoded',
		'Origin': 'https://registry.faa.gov',
		'DNT': '1',
		'Sec-GPC': '1',
		'Connection': 'keep-alive',
		'Referer': 'https://registry.faa.gov/aircraftinquiry/Search/NNumberInquiry',
		'Upgrade-Insecure-Requests': '1',
		'Sec-Fetch-Dest': 'document',
		'Sec-Fetch-Mode': 'navigate',
		'Sec-Fetch-Site': 'same-origin',
		'Sec-Fetch-User': '?1',
		'Priority': 'u=1'
	}
	data = { 'NNumbertxt': n_number }
	response = requests.post( url , headers=headers , data=data )
	response.raise_for_status()
	parsed_html = parse_aircraft_info( response.text )
	return parsed_html

if __name__ == "__main__":
	fr_api = FlightRadar24API()
	airports = airportsdata.load( "IATA" )
	call_sign_parser = FlydenityParser()
	CENTER = NEWARK
	DISTANCE = 10000
	RADIUS = 1  # Radius in kilometers for the intersection check
	bounds = fr_api.get_bounds_by_point( CENTER[ 0 ] , CENTER[ 1 ] , DISTANCE )
	flights = fr_api.get_flights( bounds=bounds )
	total_flights = len( flights )
	parsed_flights = []
	for i , flight in enumerate( flights ):
		parsed = {
			"last_update_time": flight.time ,
			"aircraft": {
				"call_sign": flight.callsign ,
				"squak": flight.squawk ,
				"code": flight.aircraft_code ,
				"country_registration": call_sign_parser.parse( flight.callsign ) ,
				"iata": flight.airline_iata ,
				"tail_number": flight.number ,
				"registration": flight.registration ,
				# "faa_registration": search_n_number( flight.registration ) ,
			} ,
			"origin": {
				"airport_iata": flight.origin_airport_iata ,
				"airport_info": None
			} ,
			"destination": {
				"airport_iata": flight.destination_airport_iata ,
				"airport_info": None
			} ,
			"trajectory": {
				"grounded": bool( flight.on_ground ) ,
				"lat": flight.latitude ,
				"lng": flight.longitude ,
				"altitude": flight.altitude ,
				"heading": flight.heading ,
				"ground_speed": flight.ground_speed ,
				"vertical_speed": flight.vertical_speed ,
			}
		}
		if parsed[ "origin" ][ "airport_iata" ] in airports:
			parsed[ "origin" ][ "airport_info" ] = airports[ parsed[ "origin" ][ "airport_iata" ] ]
		if parsed[ "destination" ][ "airport_iata" ] in airports:
			parsed[ "destination" ][ "airport_info" ] = airports[ parsed[ "destination" ][ "airport_iata" ] ]


		# Skip Currently Grounded Flights
		if parsed[ "trajectory" ][ "grounded" ] == True:
			continue
		if parsed[ "trajectory" ][ "altitude" ] < 1:
			continue

		intersects, distance, time_to_arrive, vertical_distance, heading_rad = calculate_intersection(
			CENTER, RADIUS, parsed["trajectory"]["lat"], parsed["trajectory"]["lng"],
			parsed["trajectory"]["heading"], parsed["trajectory"]["ground_speed"], parsed["trajectory"]["vertical_speed"]
		)

		if intersects == False:
			continue

		parsed["trajectory"][ "intersects" ] = intersects
		parsed["trajectory"][ "distance" ] = distance
		parsed["trajectory"][ "time_to_arrive" ] = time_to_arrive
		parsed["trajectory"][ "vertical_distance" ] = vertical_distance
		parsed["trajectory"][ "heading_rad" ] = heading_rad

		parsed_flights.append( parsed )

		# pprint( parsed )
		print( f"\n[ {i+1} ] of {total_flights}" )
		print( f"Flight: {parsed['aircraft']['call_sign']}" )
		print( f"Registration: {parsed['aircraft']['registration']}" )
		print( f"Origin: {parsed['origin']['airport_iata']}" )
		print( f"Destination: {parsed['destination']['airport_iata']}" )
		print( f"Intersects: {intersects}" )
		print( f"Distance: {distance} km" )
		print( f"Time to arrive: {time_to_arrive} hours" )
		print( f"Vertical distance: {vertical_distance} m" )


	m = plot_flight_trajectories(CENTER, RADIUS, parsed_flights)
	m.save('all_flight_trajectories.html')
	print("Map saved as all_flight_trajectories.html")