01234567890123456789012345678901234567890123456789012345678901234567890123456789
17881789179017911792179317941795179617971798179918001801180218031804180518061807 180818091810181118121813 181418151816181718181819182018211822 1823 18241825182618271828182918301831183218331834183518361837183818391840184118421843 |
<----SKIPPED LINES---->
# just simply need to continue updating this list to keep the
# dictionary up to date (i.e.: we don't need to directly touch the
# flights dictionary in main).
(last_seen, current_path) = persistent_path.get(id_to_use, (None, []))
if ( # flight position has been updated with this radio signal
not current_path or
simplified_aircraft.get('lat') != current_path[-1].get('lat') or
simplified_aircraft.get('lon') != current_path[-1].get('lon')):
current_path.append(simplified_aircraft)
persistent_path[id_to_use] = (now, current_path)
# if the flight was last seen too far in the past, remove the track info
for f in list(persistent_path.keys()):
(last_seen, current_path) = persistent_path[f]
if last_seen < now - PERSISTENCE_SECONDS:
persistent_path.pop(f)
return (nearby_aircraft, now, json_desc_dict, persistent_path)
def GetFlightAwareJson(flight_number):
"""Scrapes the text json message from FlightAware for a given flight number.
Given a flight number, loads the corresponding FlightAware webpage for that
flight and extracts the relevant script that contains all the flight details
from that page.
Args:
flight_number: text flight number (i.e.: SWA1234)
Returns:
Two tuple:
- Text representation of the json message from FlightAware.
- Text string of error message, if any
"""
url = 'https://flightaware.com/live/flight/' + flight_number
try:
response = requests.get(url)
except requests.exceptions.RequestException as e:
error_msg = 'Unable to query FA for URL due to %s: %s' % (e, url)
Log(error_msg)
return '', error_msg
soup = bs4.BeautifulSoup(response.text, 'html.parser')
l = soup.find_all('script')
flight_script = None
for script in l:
if "trackpollBootstrap" in str(script):
flight_script = str(script)
break
if not flight_script:
error_msg = (
'Unable to find trackpollBootstrap script in page: ' + response.text)
Log(error_msg)
return '', error_msg
first_open_curly_brace = flight_script.find('{')
last_close_curly_brace = flight_script.rfind('}')
<----SKIPPED LINES---->
|
01234567890123456789012345678901234567890123456789012345678901234567890123456789
17881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858 |
<----SKIPPED LINES---->
# just simply need to continue updating this list to keep the
# dictionary up to date (i.e.: we don't need to directly touch the
# flights dictionary in main).
(last_seen, current_path) = persistent_path.get(id_to_use, (None, []))
if ( # flight position has been updated with this radio signal
not current_path or
simplified_aircraft.get('lat') != current_path[-1].get('lat') or
simplified_aircraft.get('lon') != current_path[-1].get('lon')):
current_path.append(simplified_aircraft)
persistent_path[id_to_use] = (now, current_path)
# if the flight was last seen too far in the past, remove the track info
for f in list(persistent_path.keys()):
(last_seen, current_path) = persistent_path[f]
if last_seen < now - PERSISTENCE_SECONDS:
persistent_path.pop(f)
return (nearby_aircraft, now, json_desc_dict, persistent_path)
last_query_time = 0
def GetFlightAwareJson(flight_number):
"""Scrapes the text json message from FlightAware for a given flight number.
Given a flight number, loads the corresponding FlightAware webpage for that
flight and extracts the relevant script that contains all the flight details
from that page. But only queries at most once per fixed period of time
so as to avoid being blocked.
Args:
flight_number: text flight number (i.e.: SWA1234)
Returns:
Two tuple:
- Text representation of the json message from FlightAware.
- Text string of error message, if any
"""
min_query_delay_seconds = 90
global last_query_time
url = 'https://flightaware.com/live/flight/' + flight_number
seconds_since_last_query = time.time() - last_query_time
if seconds_since_last_query < min_query_delay_seconds:
error_msg = (
'Unable to query FA for URL since last query to FA was only %d seconds '
'ago; min of %d seconds needed: %s' % (
seconds_since_last_query, min_query_delay_seconds, url))
Log(error_msg)
return '', error_msg
last_query_time = time.time()
try:
response = requests.get(url)
except requests.exceptions.RequestException as e:
error_msg = 'Unable to query FA for URL due to %s: %s' % (e, url)
Log(error_msg)
return '', error_msg
soup = bs4.BeautifulSoup(response.text, 'html.parser')
l = soup.find_all('script')
flight_script = None
for script in l:
if "trackpollBootstrap" in str(script):
flight_script = str(script)
break
if not flight_script:
error_msg = (
'Unable to find trackpollBootstrap script in page: ' + response.text)
Log(error_msg)
return '', error_msg
first_open_curly_brace = flight_script.find('{')
last_close_curly_brace = flight_script.rfind('}')
<----SKIPPED LINES---->
|