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----> |