-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathham_data.py
executable file
·171 lines (160 loc) · 6.53 KB
/
ham_data.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
#!/usr/bin/env python3
import time
import logging
import pathlib
import arrow
import requests
from airport_info import get_airport_icao
from airline_info import get_airline_icao
import flight_data_source
from config import HAM_API_KEY
logger = logging.getLogger(pathlib.Path(__file__).name)
def request_ham_data():
url = "https://rest.api.hamburg-airport.de/v2/flights/"
headers = {
"Cache-Control": "no-cache",
"Ocp-Apim-Subscription-Key": HAM_API_KEY,
}
session = requests.Session()
session.headers.update(headers)
arrivals_response = session.get(url + "arrivals")
if arrivals_response.status_code != 200:
# retry once if request was not successful
time.sleep(5)
arrivals_response = session.get(url + "arrivals")
arrivals_response.raise_for_status()
arrivals = arrivals_response.json()
departures_response = session.get(url + "departures")
if departures_response.status_code != 200:
# retry once if request was not successful
time.sleep(5)
departures_response = session.get(url + "departures")
departures_response.raise_for_status()
departures = departures_response.json()
arriving_flight_numbers = [_row["flightnumber"] for _row in arrivals]
departing_flight_numbers = [_row["flightnumber"] for _row in departures]
overlapping_flight_numbers = set(arriving_flight_numbers).intersection(
departing_flight_numbers
)
return {
"arrivals": arrivals,
"departures": departures,
"overlapping_flight_numbers": overlapping_flight_numbers,
}
def _get_date_and_time(flight):
for _key in ["plannedDepartureTime", "plannedArrivalTime"]:
if flight.get(_key) is not None:
_timestamp = arrow.get(flight[_key].split("[")[0])
_date = _timestamp.format("YYYY-MM-DD")
for _key in ["expectedDepartureTime", "expectedArrivalTime"]:
if flight.get(_key) is not None:
_timestamp = arrow.get(flight[_key].split("[")[0])
return _date, _timestamp.timestamp()
class Airport(flight_data_source.FlightDataSource):
def __init__(self):
super().__init__("HAM")
def update_data(self):
data = request_ham_data()
for _flight in data["arrivals"]:
_airline_iata = _flight["airline2LCode"]
if _airline_iata is None:
continue
if not _flight["flightnumber"][3:].isdigit():
continue
_flight_number = int(_flight["flightnumber"][3:])
_airline_name = _flight["airlineName"]
if _airline_name == "SmartLynx Airlines":
_airline_iata = "6Y"
_airline_icao = get_airline_icao(
_airline_iata, _airline_name, _flight_number
)
if None in (_airline_icao, _airline_iata):
if _airline_iata != "ZZ":
logger.warning(
"operator information incomplete {}, {}, {}".format(
_airline_icao, _airline_iata, _airline_name
)
)
continue
_date, _timestamp = _get_date_and_time(_flight)
_route_items = [get_airport_icao(_flight["originAirport3LCode"])]
if _flight["viaAirport3LCode"] is not None:
_route_items.append(
get_airport_icao(_flight["viaAirport3LCode"])
)
_route_items.append("EDDH")
_route = "-".join(_route_items)
_key = "{}_{}_{}_{}".format(
_airline_iata, _flight_number, _date, _route
)
_ham_flight = {
"_id": _key,
"airline_iata": _airline_iata,
"airline_icao": _airline_icao,
"airline_name": _airline_name,
"flight_number": _flight_number,
"arrival": _timestamp,
"status": _flight["flightStatusArrival"],
"route": _route,
}
if _flight["cancelled"]:
_ham_flight["cancelled"] = True
if _flight["diverted"]:
_ham_flight["diverted"] = True
if _flight["flightnumber"] in data["overlapping_flight_numbers"]:
_ham_flight["overlap"] = True
self.update_flight(_ham_flight)
for _flight in data["departures"]:
_airline_iata = _flight["airline2LCode"]
if _airline_iata is None:
continue
if not _flight["flightnumber"][3:].isdigit():
continue
_flight_number = int(_flight["flightnumber"][3:])
_airline_name = _flight["airlineName"]
if _airline_name == "SmartLynx Airlines":
_airline_iata = "6Y"
_airline_icao = get_airline_icao(
_airline_iata, _airline_name, _flight_number
)
if None in (_airline_icao, _airline_iata):
if _airline_iata != "ZZ":
logger.warning(
"operator information incomplete {}, {}, {}".format(
_airline_icao, _airline_iata, _airline_name
)
)
continue
_date, _timestamp = _get_date_and_time(_flight)
_route_items = ["EDDH"]
if _flight["viaAirport3LCode"] is not None:
_route_items.append(
get_airport_icao(_flight["viaAirport3LCode"])
)
_route_items.append(
get_airport_icao(_flight["destinationAirport3LCode"])
)
_route = "-".join(_route_items)
_key = "{}_{}_{}_{}".format(
_airline_iata, _flight_number, _date, _route
)
_ham_flight = {
"_id": _key,
"airline_iata": _airline_iata,
"airline_icao": _airline_icao,
"airline_name": _airline_name,
"flight_number": _flight_number,
"departure": _timestamp,
"status": _flight["flightStatusDeparture"],
"route": _route,
}
if _flight["cancelled"]:
_ham_flight["cancelled"] = True
if _flight["diverted"]:
_ham_flight["diverted"] = True
if _flight["flightnumber"] in data["overlapping_flight_numbers"]:
_ham_flight["overlap"] = True
self.update_flight(_ham_flight)
if __name__ == "__main__":
airport = Airport()
airport.update_data()