-
Notifications
You must be signed in to change notification settings - Fork 4
/
Copy pathmanager.py
200 lines (173 loc) · 6.94 KB
/
manager.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
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
import glob
import json
import operator
import os
from requests import Session
from requests.auth import HTTPBasicAuth
from zeep import Client, Settings
from zeep.cache import SqliteCache
from zeep.helpers import serialize_object
from zeep.transports import Transport
import utils
from shared_vars import (DEFAULT_B2B_VERSION, DEFAULT_DATASET, WSDL_PROXY,
WSDL_PREOPS_MAIN, get_dataset)
# NOTE : plutôt envoyer un **kwargs en param des méthodes. Puis, itérer sur les kwargs.
# si présence de clé dans _default_params_for_queries et pas de val...
class Manager():
def __init__(
self,
version=DEFAULT_B2B_VERSION,
service_group="FlightServices",
dataset=DEFAULT_DATASET,
how_to_auth='proxy',
*args, **kwargs
):
self.wsdl = ''
self.service_group = service_group
self.version = version
self.session = Session()
self.available_services = []
self.available_operations = {}
self._default_params_for_queries = {
'requestedFlightFields': [],
'dataset': get_dataset(dataset),
'trafficType': 'LOAD',
'includeProposalFlights': True,
'includeForecastFlights': True,
'sendTime': utils.sendTime(),
}
self.params_for_queries = self._default_params_for_queries
# ------------- certificat ou proxy ? ------------- #
if how_to_auth == 'cert':
# Si authentification par certificat
self.wsdl = f"data/wsdl/{self.version}/{self.service_group}_PREOPS_{self.version}.wsdl"
self.session.cert = (glob.glob("cert/crt.pem")[0], glob.glob("cert/key.pem")[0])
elif how_to_auth == 'proxy':
# Si authentification via proxy (défaut)
NM_B2B_API_KEY_ID = os.environ.get('NM_B2B_API_KEY_ID') # default is None
NM_B2B_API_SECRET = os.environ.get('NM_B2B_API_SECRET') # default is None
if not NM_B2B_API_KEY_ID or not NM_B2B_API_SECRET:
print(f"Impossible de définir un couple clé/pass pour le proxy b2b.\
Vérifiez que NM_B2B_API_KEY_ID et NM_B2B_API_SECRET sont bien définis dans votre environnement.")
exit(1)
self.wsdl = WSDL_PROXY + self.service_group + "_PREOPS_" + version + ".wsdl"
self.session.auth = HTTPBasicAuth(NM_B2B_API_KEY_ID, NM_B2B_API_SECRET)
else:
print("Le mode d'authentification que vous avez spécifié n'existe pas (pour l'instant : 'cert' ou 'proxy').")
exit(1)
# -------------
self.cache = SqliteCache(path='./data/sqlite.db')
self.transport = Transport(session=self.session, cache=self.cache)
self.conf = {
'wsdl': self.wsdl,
'transport': self.transport
}
self.tmp_data = None
# -----------------------------------------------------------------------------------------
def set_available_services(self):
self.available_services = []
client = Client(**self.conf)
for service in client.wsdl.services.values():
self.available_services.append(service.__str__().split()[1])
def show_available_services(self):
if not self.available_services:
self.set_available_services()
print("Les services disponibles via {} sont : ".format(self.service_group)),
for service in self.available_services:
print(' * ', service)
# -----------------------------------------------------------------------------------------
def set_operations_of_service(self, service_name):
if not self.available_services:
self.set_available_services()
if not service_name in self.available_services:
raise Exception(f"Le service {service_name} n'est pas disponible.")
client = Client(**self.conf, service_name=service_name)
operations = [op for op in client.service.__dir__() if not op.startswith('__')]
self.available_operations[service_name] = operations
def show_operations_of_service(self, service_name):
if not service_name in self.available_operations:
self.set_operations_of_service(service_name=service_name)
print(f"Les opérations disponibles pour {service_name} sont : ")
for operation in self.available_operations[service_name]:
print(' * ', operation)
# -----------------------------------------------------------------------------------------
def convert_data_to_json(self, obj):
return serialize_object(self.tmp_data)
def update_params_for_queries(self, new_params):
for param in new_params:
if param in self.params_for_queries:
self.params_for_queries[param] = new_params[param]
def set_traffic_window(self, startTime, endTime):
return {
'wef': utils.format_datetime_for_nm(startTime),
'unt': utils.format_datetime_for_nm(endTime)
}
# -----------------------------------------------------------------------------------------
def queryFlightsByAirspace(self, airspace, startTime, endTime, other_params={}):
""" Liste des vols par airspace
Récupère la liste des vols transitant par un volume d'espace donné.
Parameters
----------
airspace
volume d'espace souhaité
startTime
heure de début (UTC)
endTime
heure de fin (UTC)
other_params
dictionnaire pour overrider les valeurs par défaut de _default_params_for_queries
"""
trafficWindow = self.set_traffic_window(startTime, endTime)
self.update_params_for_queries(other_params)
client = Client(**self.conf, service_name='FlightManagementService')
self.tmp_data = client.service.queryFlightsByAirspace(
airspace=airspace,
trafficWindow=trafficWindow,
**self.params_for_queries)
return self.tmp_data
def queryFlightsByAerodrome(self, aerodrome, aerodromeRole, startTime, endTime, other_params={}):
""" Liste des vols par aérodrome
Récupère la liste des vols au départ ou à l'arrivée d'un aérodrome.
Parameters
----------
aerodrome
code OACI du terrain souhaité
aerodromeRole
rôle du terrain (DEPARTURE...)
startTime
heure de début (UTC)
endTime
heure de fin (UTC)
other_params
dictionnaire pour overrider les valeurs par défaut de _default_params_for_queries
"""
trafficWindow = self.set_traffic_window(startTime, endTime)
self.update_params_for_queries(other_params)
client = Client(**self.conf, service_name='FlightManagementService')
self.tmp_data = client.service.queryFlightsByAerodrome(
aerodrome=aerodrome, aerodromeRole=aerodromeRole,
trafficWindow=trafficWindow,
**self.params_for_queries)
return self.tmp_data
def queryFlightsByTrafficVolume(self, trafficVolume, startTime, endTime, other_params={}):
""" Liste des vols par Traffic Volume (TV)
Récupère la liste des vols transitant par un TV.
Parameters
----------
trafficVolume
nom du TV
startTime
heure de début (UTC)
endTime
heure de fin (UTC)
other_params
dictionnaire pour overrider les valeurs par défaut de _default_params_for_queries
"""
trafficWindow = self.set_traffic_window(startTime, endTime)
self.update_params_for_queries(other_params)
client = Client(**self.conf, service_name='FlightManagementService')
self.tmp_data = client.service.queryFlightsByTrafficVolume(
trafficVolume=trafficVolume,
trafficWindow=trafficWindow,
**self.params_for_queries)
return self.tmp_data