-
Notifications
You must be signed in to change notification settings - Fork 1
/
tfl.py
200 lines (159 loc) · 7.98 KB
/
tfl.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
from telegram import ForceReply, Update
from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters
import daisySecrets
import datetime, requests, time
async def tflstatus_command(update: Update, context: ContextTypes.DEFAULT_TYPE, mode="tfl") -> None:
"""Send the status of transit using TfL APIs."""
# TfL line to get the status for
message_args = update.message.text.split(' ')
if (len(message_args) == 1) or (message_args[1] == "nr"):
try:
if message_args[1] == "nr":
service = "nr"
all_lines_status_url = 'https://api.tfl.gov.uk/line/mode/overground,elizabeth-line,national-rail/status'
except:
service = "tfl"
all_lines_status_url = 'https://api.tfl.gov.uk/line/mode/tube,overground,elizabeth-line,dlr,tram/status'
# Make a request to the TfL API to get the status of all lines
response = requests.get(all_lines_status_url)
# Check if the request was successful
if response.status_code == requests.codes.ok:
# Parse the response as JSON
all_lines_status = response.json()
# Iterate over each line and display the current status
if service == "nr":
status_output = "<b>Current status on all National Rail services</b>"
else:
status_output = "<b>Current status on all TfL services</b>"
for line in all_lines_status:
disruptions = line['lineStatuses']
if disruptions:
lineName = fixLineName(line['name'])
status = disruptions[0]['statusSeverityDescription']
status_output += "\n" + lineName + ": " + status
await update.message.reply_html(f"{status_output}")
else:
await update.message.reply_text(f"Error retrieving line status: {response.status_code}")
else:
# TfL API endpoint for getting the line status
line_status_url = 'https://api.tfl.gov.uk/line/{line}/status'
line_id = message_args[1]
if line_id == "overground":
line_id = "london-overground"
elif line_id == "gwr":
line_id = "great-western-railway"
elif line_id == "trams":
line_id = "tram"
elif line_id == "swr":
line_id = "south-western-railway"
# Make a request to the TfL API to get the line status
response = requests.get(line_status_url.format(line=line_id))
# Check if the request was successful
if response.status_code == requests.codes.ok:
# Parse the response as JSON
line_status = response.json()[0]['lineStatuses'][0]
line_status_summary = line_status['statusSeverityDescription']
replyText = "Currently "
isTube = (response.json()[0]["modeName"] == "tube")
if isTube or response.json()[0]["modeName"] == "elizabeth-line" or response.json()[0]["modeName"] == "tram": # Thanks Boris for this terrible name
replyText += "the "
lineName = fixLineName(response.json()[0]['name'])
replyText += lineName
if isTube:
replyText += " line"
if line_status_summary == "Good Service":
replyText += " has a good service."
elif line_status_summary == "Part Suspended":
replyText += " is partially suspended."
elif line_status_summary == "Part Closure":
replyText += " is partially closed."
elif line_status_summary == "Planned Closure":
replyText += " is closed."
else:
replyText += f" has {line_status_summary}."
if 'reason' in line_status:
replyText = replyText + "\n\n" + line_status["reason"]
await update.message.reply_text(replyText)
else:
await update.message.reply_text(f"Error retrieving line status: {response.status_code}")
def fixLineName(name):
if name == "Tram":
name = "Trams"
return name
async def tflbus_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
message_args = update.message.text.split(' ')
# Replace 'STOP_CODE' with the 5-digit bus stop code you want to retrieve data for
try:
stop_code = int(message_args[1])
except:
await update.message.reply_text("Failed to read stop code")
return
try:
mode = message_args[2]
mode = mode.lower()
if mode == "-v" or mode == "v":
mode = "v"
except:
mode = "n"
# Base URL for the TFL Unified API
base_url = 'https://api.tfl.gov.uk'
# Endpoint for the Search API to resolve the stop code to Naptan ID
search_url = f'{base_url}/StopPoint/Search?query={stop_code}'
# Make the API request to resolve the stop code to Naptan ID
response = requests.get(search_url)
# Check if the request was successful
if response.status_code == 200:
# Parse the JSON response
stop_points = response.json()
# Extract the first stop point (assuming exact match)
if len(stop_points) > 0:
stop_point = stop_points['matches'][0]
naptan_id = stop_point['id']
# Endpoint for the StopPoint API with the resolved Naptan ID
stoppoint_url = f'{base_url}/StopPoint/{naptan_id}/Arrivals'
# Make the API request to retrieve bus times
response = requests.get(stoppoint_url)
if response.status_code == 200:
# Parse the JSON response
bus_times = response.json()
if len(bus_times) == 0:
await update.message.reply_text(f"No bus times available for {stop_point['name']}")
return
replyText = f"Next buses for {stop_point['name']}"
try:
replyText += f" Stop {stop_point['stopLetter']} \n"
except:
replyText += "\n"
sortedBus = {}
# Extract relevant information
for bus in bus_times:
bus_id = bus['vehicleId']
destination = bus['destinationName']
line = bus['lineName']
try:
expected_arrival = int(time.strftime('%M', time.gmtime(bus['timeToStation'])))
except:
try:
arrival_time = datetime.fromisoformat(bus['expectedArrival'])
current_time = datetime.utcnow()
expected_arrival = int((arrival_time - current_time).total_seconds() / 60)
except:
expected_arrival = 0
sortedBus[bus_id] = {"arrival": expected_arrival, "destination": destination, "line": line}
sortedBus = dict(sorted(sortedBus.items(), key=lambda item: item[1]["arrival"]))
for bus in sortedBus:
if mode == "v":
replyText += "\n" + str(sortedBus[bus]["arrival"]) + " minutes " + sortedBus[bus]["line"] + " " + sortedBus[bus]["destination"] + " (" + bus + ")"
else:
replyText += "\n" + str(sortedBus[bus]["arrival"]) + " minutes " + sortedBus[bus]["line"] + " " + sortedBus[bus]["destination"]
await update.message.reply_text(replyText)
return
else:
await update.message.reply_text('Error occurred while retrieving bus times.')
return
else:
await update.message.reply_text('No stop point found for the specified stop code.')
return
else:
await update.message.reply_text('Error occurred while resolving the stop code to Naptan ID.')
return