01234567890123456789012345678901234567890123456789012345678901234567890123456789
15331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573 19171918191919201921192219231924192519261927192819291930193119321933193419351936 193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960 19611962196319641965196619671968 1969197019711972 19731974197519761977197819791980198119821983198419851986198719881989199019911992 |
<----SKIPPED LINES---->
are a sequential list of the location-attributes in the json file; allows
for tracking the flight path over time.
log_jsons: boolean indicating whether we should pickle the JSONs.
flights: list of flight dictionaries; if no json is returned, used to
find a recent flight with same flight number to augment this flight with
origin / destination / airline.
Returns:
A tuple:
- updated persistent_nearby_aircraft
- (possibly empty) dictionary of flight attributes of the new flight upon
its first observation.
- the time of the radio observation if present; None if no radio dump
- a dictionary of attributes about the dump itself (i.e.: # of flights;
furthest observed flight, etc.)
- persistent_path, a data structure containing past details of a flight's
location as described in ParseDumpJson
- a text message indicating any errors in querying FlightAware or
populating flight details
- text string of SUCCESS, WARNING, or FAILURE: warning meaning the query
occurred to soon after the last FA request, FAILURE if the request failed
for some other reason, and SUCCESS if it was otherwise successful.
- timestamp indicating exact time at which FlightAware was queried (or
attempted to be queried), if a query was made in this pass
"""
flight_details = {}
error_message = ''
status = 'SUCCESS'
now = time.time()
if SIMULATION:
(dump_json, json_time) = DUMP_JSONS[SIMULATION_COUNTER]
else:
dump_json = ReadFile(DUMP_JSON_FILE, log_exception=True)
# Often there is no flight aware query, so we need to give a default value
flight_aware_timestamp = 0
json_desc_dict = {}
current_nearby_aircraft = {}
if dump_json:
(current_nearby_aircraft, now,
json_desc_dict, persistent_path) = ParseDumpJson(
<----SKIPPED LINES---->
# 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:
Four tuple:
- Text representation of the json message from FlightAware.
- Text string of error message, if any
- Text string of SUCCESS, WARNING, or FAILURE: warning meaning the query
occurred to soon after the last FA request, FAILURE if the request failed
for some other reason, and SUCCESS if it was otherwise successful.
- Timestamp of attempted query on FlightAware
"""
min_query_delay_seconds = 90
url = 'https://flightaware.com/live/flight/' + flight_number
global last_query_time
seconds_since_last_query = time.time() - last_query_time
if last_query_time and seconds_since_last_query < min_query_delay_seconds:
error_msg = (
'Unable to query FA for %s at %s since last query to FA was only'
' %d seconds ago; min of %d seconds needed: %s' % (
flight_number,
EpochDisplayTime(time.time(), format_string='%H:%M:%S'),
seconds_since_last_query, min_query_delay_seconds, url))
flight_aware_status_code = 'WARNING'
return '', error_msg, flight_aware_status_code, time.time()
last_query_time = time.time()
try:
response = requests.get(url, timeout=5)
query_time = time.time()
except requests.exceptions.RequestException as e:
query_time = time.time() # did not get to the query_time assignment above
error_msg = 'Unable to query FA for URL due to %s: %s' % (e, url)
flight_aware_status_code = 'FAILURE'
return '', error_msg, flight_aware_status_code, query_time
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)
<----SKIPPED LINES---->
|
01234567890123456789012345678901234567890123456789012345678901234567890123456789
15331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573 1917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998 |
<----SKIPPED LINES---->
are a sequential list of the location-attributes in the json file; allows
for tracking the flight path over time.
log_jsons: boolean indicating whether we should pickle the JSONs.
flights: list of flight dictionaries; if no json is returned, used to
find a recent flight with same flight number to augment this flight with
origin / destination / airline.
Returns:
A tuple:
- updated persistent_nearby_aircraft
- (possibly empty) dictionary of flight attributes of the new flight upon
its first observation.
- the time of the radio observation if present; None if no radio dump
- a dictionary of attributes about the dump itself (i.e.: # of flights;
furthest observed flight, etc.)
- persistent_path, a data structure containing past details of a flight's
location as described in ParseDumpJson
- a text message indicating any errors in querying FlightAware or
populating flight details
- text string of SUCCESS, WARNING, or FAILURE: warning meaning the query
occurred too soon after the last FA request, FAILURE if the request failed
for some other reason, and SUCCESS if it was otherwise successful.
- timestamp indicating exact time at which FlightAware was queried (or
attempted to be queried), if a query was made in this pass
"""
flight_details = {}
error_message = ''
status = 'SUCCESS'
now = time.time()
if SIMULATION:
(dump_json, json_time) = DUMP_JSONS[SIMULATION_COUNTER]
else:
dump_json = ReadFile(DUMP_JSON_FILE, log_exception=True)
# Often there is no flight aware query, so we need to give a default value
flight_aware_timestamp = 0
json_desc_dict = {}
current_nearby_aircraft = {}
if dump_json:
(current_nearby_aircraft, now,
json_desc_dict, persistent_path) = ParseDumpJson(
<----SKIPPED LINES---->
# 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
last_query_flight_number = ''
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:
Four tuple:
- Text representation of the json message from FlightAware.
- Text string of error message, if any
- Text string of SUCCESS, WARNING, or FAILURE: warning meaning the query
occurred too soon after the last FA request, FAILURE if the request
failed for some other reason, and SUCCESS if it was otherwise successful.
- Timestamp of attempted query on FlightAware
"""
min_query_delay_seconds = 90
url = 'https://flightaware.com/live/flight/' + flight_number
global last_query_time
global last_query_flight_number
seconds_since_last_query = time.time() - last_query_time
if last_query_time and seconds_since_last_query < min_query_delay_seconds:
error_msg = (
'Unable to query FA for %s at %s since last query to FA was only'
' %d seconds ago for %s; min of %d seconds needed: %s' % (
flight_number,
EpochDisplayTime(time.time(), format_string='%H:%M:%S'),
seconds_since_last_query,
last_query_flight_number,
min_query_delay_seconds,
url))
flight_aware_status_code = 'WARNING'
return '', error_msg, flight_aware_status_code, time.time()
last_query_time = time.time()
last_query_flight_number = flight_number
try:
response = requests.get(url, timeout=5)
query_time = time.time()
except requests.exceptions.RequestException as e:
query_time = time.time() # did not get to the query_time assignment above
error_msg = 'Unable to query FA for URL due to %s: %s' % (e, url)
flight_aware_status_code = 'FAILURE'
return '', error_msg, flight_aware_status_code, query_time
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)
<----SKIPPED LINES---->
|