-
Notifications
You must be signed in to change notification settings - Fork 0
/
time-fc.py
executable file
·165 lines (133 loc) · 7.35 KB
/
time-fc.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
import os
import json
import redis
from flask import Flask, request
from datetime import datetime as dt
from datetime import timedelta
import configparser as cp
import logging
from r2r_offer_utils import normalization
from r2r_offer_utils.cache_operations import read_data_from_cache_wrapper, store_simple_data_to_cache_wrapper
from r2r_offer_utils.logging import setup_logger
import rush_hours
service_name = os.path.splitext(os.path.basename(__file__))[0]
app = Flask(service_name)
# config
config = cp.ConfigParser()
config.read(f'{service_name}.conf')
# cache
cache = redis.Redis(host=config.get('cache', 'host'),
port=config.get('cache', 'port'),
decode_responses=True)
norm_type = config.get('running', 'scores')
# init logging
logger, ch = setup_logger()
logger.setLevel(logging.INFO)
@app.route('/compute', methods=['POST'])
def extract():
data = request.get_json()
request_id = data['request_id']
# ask for the entire list of offer ids
offer_data = cache.lrange('{}:offers'.format(request_id), 0, -1)
response = app.response_class(
response=f'{{"request_id": "{request_id}"}}',
status=200,
mimetype='application/json'
)
output_offer_level, output_tripleg_level = read_data_from_cache_wrapper(pa_cache=cache, pa_request_id=request_id,
pa_offer_level_items=['start_time', 'end_time'],
pa_tripleg_level_items=['start_time', 'end_time'])
# to be replaced with the actual date and time of the request
offer_start_time_string = output_offer_level[output_offer_level['offer_ids'][0]]['start_time']
try:
offer_start_time = dt.fromisoformat(offer_start_time_string)
except ValueError:
# this is to handle an error in the formatting of the time string in some TRIAS files
offer_start_time_string = offer_start_time_string[:offer_start_time_string.index('+')] + '0' + offer_start_time_string[offer_start_time_string.index('+'):]
offer_start_time = dt.fromisoformat(offer_start_time_string)
# get the time zone from one of the leg_times, or else default it to UTC
try:
time_zone = offer_start_time.tzinfo
except:
time_zone = timezone.utc
current_time = dt.now(tz=time_zone)
logger.info(f'Current time: {current_time}')
offer_features = {'duration':{},
'time_to_departure':{},
'waiting_time':{},
'rush_overlap':{}}
for offer_id in output_offer_level['offer_ids']:
offer_start_time_string = output_offer_level[offer_id]['start_time']
try:
offer_start_time = dt.fromisoformat(offer_start_time_string)
except ValueError:
# this is to handle an error in the formatting of the time string in some TRIAS files
offer_start_time_string = offer_start_time_string[:offer_start_time_string.index('+')] + '0' + offer_start_time_string[offer_start_time_string.index('+'):]
offer_start_time = dt.fromisoformat(offer_start_time_string)
logger.info(f'Offer {offer_id} starts at {offer_start_time}')
offer_end_time_string = output_offer_level[offer_id]['end_time']
try:
offer_end_time = dt.fromisoformat(offer_end_time_string)
except ValueError:
# this is to handle an error in the formatting of the time string in some TRIAS files
offer_end_time_string = offer_end_time_string[:offer_end_time_string.index('+')] + '0' + offer_end_time_string[offer_end_time_string.index('+'):]
offer_end_time = dt.fromisoformat(offer_end_time_string)
offer_duration = (offer_end_time - offer_start_time).total_seconds()/60
offer_time_to_departure = (offer_start_time - current_time).total_seconds()/60
rush_minutes, rush_overlap = rush_hours.calc_rush_overlap(offer_start_time, offer_end_time, country='default')
# compute the overall waiting time between legs
leg_ids = list(reversed((output_tripleg_level[offer_id]['triplegs'])))
waiting_time_between_legs = 0
if len(leg_ids) > 1:
for i in range(1, len(leg_ids)):
previous_end_time_string = output_tripleg_level[offer_id][leg_ids[i-1]]['end_time']
try:
previous_end_time = dt.fromisoformat(previous_end_time_string)
except ValueError:
# this is to handle an error in the formatting of the time string in some TRIAS files
previous_end_time_string = previous_end_time_string[:previous_end_time_string.index('+')] + '0' + previous_end_time_string[previous_end_time_string.index('+'):]
previous_end_time = dt.fromisoformat(previous_end_time_string)
next_start_time_string = output_tripleg_level[offer_id][leg_ids[i]]['start_time']
try:
next_start_time = dt.fromisoformat(next_start_time_string)
except ValueError:
# this is to handle an error in the formatting of the time string in some TRIAS files
next_start_time_string = next_start_time_string[:next_start_time_string.index('+')] + '0' + next_start_time_string[next_start_time_string.index('+'):]
next_start_time = dt.fromisoformat(next_start_time_string)
waiting_time_between_legs += (next_start_time - previous_end_time).total_seconds()/60
logger.info(f'Time to departure: {offer_time_to_departure}')
logger.info(f'Waiting time between legs: {waiting_time_between_legs}')
offer_features['duration'][offer_id] = offer_duration
if offer_time_to_departure < 0:
negative_time_to_departure = True
offer_time_to_departure *= -1
else:
negative_time_to_departure = False
offer_features['time_to_departure'][offer_id] = offer_time_to_departure
offer_features['waiting_time'][offer_id] = waiting_time_between_legs
offer_features['rush_overlap'][offer_id] = rush_overlap
# normalize features and store them to cache
offer_features_norm = {}
for feature in offer_features:
if feature == 'time_to_departure' and negative_time_to_departure:
flipped = False
else:
flipped = True
if norm_type == 'z_score':
offer_features_norm[feature] = normalization.zscore(offer_features[feature], flipped=flipped)
else:
offer_features_norm[feature] = normalization.minmaxscore(offer_features[feature], flipped=flipped)
logger.info(f'{feature}, {offer_features_norm[feature]}')
store_simple_data_to_cache_wrapper(pa_cache=cache,
pa_request_id=request_id,
pa_data=offer_features_norm[feature],
pa_sub_key=feature)
return response
if __name__ == '__main__':
FLASK_PORT = 5000
REDIS_HOST = 'localhost'
REDIS_PORT = 6379
os.environ["FLASK_ENV"] = "development"
cache = redis.Redis(host=REDIS_HOST, port=REDIS_PORT)
#print(cache.keys())
app.run(port=FLASK_PORT, debug=True)