-
Notifications
You must be signed in to change notification settings - Fork 0
/
traintimes.py
146 lines (109 loc) · 5.12 KB
/
traintimes.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
# Author: Joel Ruuben Seene
"""
Project idea: Query train times from Elron (Estonian National Passenger Rail Provider)
Ask starting stop and destination stop, date --> Check the list of stops for stop id (internal code)
If the stop doesn't exist, notify the user
Make query against the API for schedule
Filter out necessary data
present as table in console
Possible further development --> do the same for LuxExpress site
Necessity with Lux --> Check for existing stop, translate to code, create URL, make query, etc
Show both side-by-side --> Elron | LuxExpress (if applicable)
"""
from json import load, dumps
from datetime import date, datetime, timezone, timedelta
from requests import request
from tabulate import tabulate
def stop_id(stop_location):
try:
a = stop_location.strip()
b = a.capitalize()
with open("Stops.json", encoding="utf8") as f:
data = load(f)
for stops in data:
if stops["stop_name"] == b:
return stops["stop_area_id"]
except TypeError as e:
print("Incorrect input")
def api_query(origin_stop_id, destination_stop_id, query_date):
"""
Queries the Ridango API for the train times between 2 stops
:param origin_stop_id: Origin Stop ID
:param destination_stop_id: Destination stop ID
:param query_date: Trip date
:return: API response in JSON
Original code generated with Postman, modified to fit the project
"""
if query_date == '': # If the input is empty, add current date (YYYY-MM-DD)
query_date = date.today()
url = "https://api.ridango.com/v2/64/intercity/stopareas/trips/direct"
payload = dumps({
"date": str(query_date),
"origin_stop_area_id": str(origin_stop_id),
"destination_stop_area_id": str(destination_stop_id),
"channel": "web"
})
headers = {
'authority': 'api.ridango.com',
'accept': 'application/json, text/plain, */*',
'accept-language': 'en-US,en;q=0.9,et;q=0.8',
'cache-control': 'no-cache',
'content-type': 'application/json',
'origin': 'https://elron.pilet.ee',
'pragma': 'no-cache',
'referer': 'https://elron.pilet.ee/',
'sec-ch-ua': '"Not.A/Brand";v="8", "Chromium";v="114", "Google Chrome";v="114"',
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'cross-site',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
'Chrome/114.0.0.0 Safari/537.36'
}
response = request("PUT", url, headers=headers, data=payload).json()
return response
"""
Function to process the API response
:param response: Response from the API in JSON
:param time: "Current time"
:return: Trip info as lists
"""
def get_trip_info(response, time=None):
output = []
for journey in response["journeys"]:
for trip in journey["trips"]:
arrival_time = datetime.fromisoformat(trip["arrival_time"]).replace(tzinfo=timezone.utc)
departure_time = datetime.fromisoformat(trip["departure_time"]).replace(tzinfo=timezone.utc)
price = str(trip["product"]["price"]) + '\u20AC' # Get the price info, add '€' as unicode
if time is not None:
if time < departure_time: # If a time to compare to has been supplied and departure_time is after it
formatted_departure = datetime.strftime(departure_time, '%H:%M') \
+ ' - ' \
+ datetime.strftime(arrival_time, '%H:%M')
departure = [formatted_departure, price]
output.append(departure)
else: # Show all, since no time to start from has been given
formatted_departure = datetime.strftime(departure_time, '%H:%M') \
+ ' - ' \
+ datetime.strftime(arrival_time, '%H:%M')
departure = [formatted_departure, price]
output.append(departure)
return output
if __name__ == '__main__':
origin = input("Input starting stop: ")
destination = input("Input destination: ")
input_date = input("Input date as YYYY-MM-DD (leave empty for today): ")
if input_date == '':
show_all_options = input("Show all trips today? (Y/N) ").capitalize()
origin_id = stop_id(origin) # "E.g. Tartu - 64-5924-93"
destination_id = stop_id(destination) # "E.g. Tallinn - 64-5924-97"
api_response = api_query(origin_id, destination_id, input_date)
current_time = datetime.now().replace(tzinfo=timezone.utc)
if input_date == '' and show_all_options == 'Y':
result = get_trip_info(api_response)
elif input_date == '' and show_all_options == 'N':
result = get_trip_info(api_response, current_time)
else:
future_date = datetime.strptime(input_date, "%Y-%m-%d").replace(tzinfo=timezone.utc)
result = get_trip_info(api_response, future_date)
print(f"\n{origin} --> {destination}")
print(tabulate(result, tablefmt="grid"))