aboutsummaryrefslogtreecommitdiff
path: root/modules/tfl.py
diff options
context:
space:
mode:
authorGravatar dngfx2018-08-31 10:50:37 +0100
committerGravatar dngfx2018-08-31 10:50:37 +0100
commitabed9cf4ea71dcbad2dd2c049683b8d14b942e09 (patch)
tree3e40caf63fa7e1500469f4ad9a0c45c51808aad4 /modules/tfl.py
parentFix a copy paste fail in IRCLineHandler that caused PARTs to be handled as QUITs (diff)
signature
Reformat
Diffstat (limited to 'modules/tfl.py')
-rw-r--r--modules/tfl.py171
1 files changed, 105 insertions, 66 deletions
diff --git a/modules/tfl.py b/modules/tfl.py
index 2a017f1d..5945bd04 100644
--- a/modules/tfl.py
+++ b/modules/tfl.py
@@ -7,7 +7,9 @@ URL_BUS_SEARCH = "https://api.tfl.gov.uk/StopPoint/Search/%s"
URL_LINE_ARRIVALS = "https://api.tfl.gov.uk/Line/%s/Arrivals"
URL_LINE = "https://api.tfl.gov.uk/Line/Mode/tube/Status"
-LINE_NAMES = ["bakerloo", "central", "circle", "district", "hammersmith and city", "jubilee", "metropolitan", "piccadilly", "victoria", "waterloo and city"]
+LINE_NAMES = ["bakerloo", "central", "circle", "district",
+ "hammersmith and city", "jubilee", "metropolitan", "piccadilly",
+ "victoria", "waterloo and city"]
URL_STOP = "https://api.tfl.gov.uk/StopPoint/%s"
URL_STOP_SEARCH = "https://api.tfl.gov.uk/StopPoint/Search/%s"
@@ -16,52 +18,62 @@ URL_VEHICLE = "https://api.tfl.gov.uk/Vehicle/%s/Arrivals"
URL_ROUTE = "https://api.tfl.gov.uk/Line/%s/Route/Sequence/all?excludeCrowding=True"
-PLATFORM_TYPES = ["Northbound", "Southbound", "Eastbound", "Westbound", "Inner Rail", "Outer Rail"]
+PLATFORM_TYPES = ["Northbound", "Southbound", "Eastbound", "Westbound",
+ "Inner Rail", "Outer Rail"]
+
class Module(object):
_name = "TFL"
+
def __init__(self, bot):
self.bot = bot
self.result_map = {}
bot.events.on("received").on("command").on("tflbus"
- ).hook(self.bus, min_args=1,
- help="Get bus due times for a TfL bus stop",
- usage="<stop_id>")
+ ).hook(self.bus, min_args=1,
+ help="Get bus due times for a TfL bus stop",
+ usage="<stop_id>")
bot.events.on("received").on("command").on("tflline"
- ).hook(self.line,
- help="Get line status for TfL underground lines",
- usage="<line_name>")
+ ).hook(self.line,
+ help="Get line status for TfL underground lines",
+ usage="<line_name>")
bot.events.on("received").on("command").on("tflsearch"
- ).hook(self.search, min_args=1,
- help="Get a list of TfL stop IDs for a given name",
- usage="<name>")
+ ).hook(self.search,
+ min_args=1,
+ help="Get a list of TfL stop IDs for a given name",
+ usage="<name>")
bot.events.on("received").on("command").on("tflvehicle"
- ).hook(self.vehicle, min_args=1,
- help="Get information for a given vehicle",
- usage="<ID>")
+ ).hook(self.vehicle,
+ min_args=1,
+ help="Get information for a given vehicle",
+ usage="<ID>")
bot.events.on("received").on("command").on("tflstop"
- ).hook(self.stop, min_args=1,
- help="Get information for a given stop",
- usage="<stop_id>")
+ ).hook(self.stop, min_args=1,
+ help="Get information for a given stop",
+ usage="<stop_id>")
bot.events.on("received").on("command").on("tflservice"
- ).hook(self.service, min_args=1,
- help="Get service information and arrival estimates",
- usage="<service index>")
+ ).hook(self.service,
+ min_args=1,
+ help="Get service information and arrival estimates",
+ usage="<service index>")
def vehicle_span(self, arrival_time, human=True):
vehicle_due_iso8601 = arrival_time
if "." in vehicle_due_iso8601:
- vehicle_due_iso8601 = vehicle_due_iso8601.split(".")[0]+"Z"
+ vehicle_due_iso8601 = vehicle_due_iso8601.split(".")[0] + "Z"
vehicle_due = datetime.datetime.strptime(vehicle_due_iso8601,
- "%Y-%m-%dT%H:%M:%SZ")
- time_until = vehicle_due-datetime.datetime.utcnow()
- time_until = int(time_until.total_seconds()/60)
+ "%Y-%m-%dT%H:%M:%SZ")
+ time_until = vehicle_due - datetime.datetime.utcnow()
+ time_until = int(time_until.total_seconds() / 60)
- if time_until == 0: human_time = "due"
- else: human_time = "in %s min" % time_until
+ if time_until == 0:
+ human_time = "due"
+ else:
+ human_time = "in %s min" % time_until
- if human: return human_time
- else: return time_until
+ if human:
+ return human_time
+ else:
+ return time_until
def platform(self, platform, short=False):
p = re.compile("(?:(.*) - Platform (\\d+)|(.*bound) Platform (\\d+))")
@@ -103,16 +115,23 @@ class Module(object):
for bus in bus_stop:
bus_number = bus["lineName"]
human_time = self.vehicle_span(bus["expectedArrival"])
- time_until = self.vehicle_span(bus["expectedArrival"], human=False)
+ time_until = self.vehicle_span(bus["expectedArrival"],
+ human=False)
# If the mode is "tube", "Underground Station" is redundant
destination = bus.get("destinationName", "?")
- if (bus["modeName"] == "tube"): destination = destination.replace(" Underground Station", "")
+ if (bus[
+ "modeName"] == "tube"): destination = destination.replace(
+ " Underground Station", "")
- busses.append({"route": bus_number, "time": time_until, "id": bus["vehicleId"],
- "destination": destination, "human_time": human_time, "mode": bus["modeName"],
- "platform": bus["platformName"],
- "platform_short" : self.platform(bus["platformName"], short=True)})
+ busses.append({"route": bus_number, "time": time_until,
+ "id": bus["vehicleId"],
+ "destination": destination,
+ "human_time": human_time,
+ "mode": bus["modeName"],
+ "platform": bus["platformName"],
+ "platform_short": self.platform(
+ bus["platformName"], short=True)})
if busses:
busses = sorted(busses, key=lambda b: b["time"])
busses_filtered = []
@@ -122,11 +141,14 @@ class Module(object):
# dedup if target route isn't "*", filter if target route isn't None or "*"
for b in busses:
if target_bus_route != "*":
- if (b["route"], b["destination"]) in bus_route_dest: continue
- if bus_route_plat.count((b["route"], b["platform"])) >= 2: continue
+ if (b["route"],
+ b["destination"]) in bus_route_dest: continue
+ if bus_route_plat.count(
+ (b["route"], b["platform"])) >= 2: continue
bus_route_plat.append((b["route"], b["platform"]))
bus_route_dest.append((b["route"], b["destination"]))
- if b["route"] == target_bus_route or not target_bus_route:
+ if b[
+ "route"] == target_bus_route or not target_bus_route:
busses_filtered.append(b)
else:
busses_filtered.append(b)
@@ -134,32 +156,33 @@ class Module(object):
self.result_map[event["target"].id] = busses_filtered
# do the magic formatty things!
- busses_string = ", ".join(["%s (%s, %s)" % (b["destination"], b["route"], b["human_time"],
- ) for b in busses_filtered])
+ busses_string = ", ".join(["%s (%s, %s)" % (
+ b["destination"], b["route"], b["human_time"],
+ ) for b in busses_filtered])
event["stdout"].write("%s (%s): %s" % (stop_name, stop_id,
- busses_string))
+ busses_string))
else:
event["stderr"].write("%s: No busses due" % stop_id)
else:
- event["stderr"].write("Bus ID '%s' unknown" % stop_id)
+ event["stderr"].write("Bus ID '%s' unknown" % stop_id)
def line(self, event):
app_id = self.bot.config["tfl-api-id"]
app_key = self.bot.config["tfl-api-key"]
lines = Utils.get_url(URL_LINE, get_params={
- "app_id": app_id, "app_key": app_key}, json=True)
+ "app_id": app_id, "app_key": app_key}, json=True)
statuses = []
for line in lines:
for status in line["lineStatuses"]:
entry = {
- "id": line["id"],
- "name": line["name"],
- "severity": status["statusSeverity"],
- "description": status["statusSeverityDescription"],
- "reason": status.get("reason")
- }
+ "id": line["id"],
+ "name": line["name"],
+ "severity": status["statusSeverity"],
+ "description": status["statusSeverityDescription"],
+ "reason": status.get("reason")
+ }
statuses.append(entry)
statuses = sorted(statuses, key=lambda line: line["severity"])
combined = collections.OrderedDict()
@@ -177,7 +200,9 @@ class Module(object):
for status in statuses:
for arg in event["args_split"]:
if arg.lower() in status["name"].lower():
- result += "%s: %s (%d) '%s'; " % (status["name"], status["description"], status["severity"], status["reason"])
+ result += "%s: %s (%d) '%s'; " % (
+ status["name"], status["description"],
+ status["severity"], status["reason"])
if result:
event["stdout"].write(result[:-2])
else:
@@ -187,16 +212,20 @@ class Module(object):
app_id = self.bot.config["tfl-api-id"]
app_key = self.bot.config["tfl-api-key"]
- #As awful as this is, it also makes it ~work~.
+ # As awful as this is, it also makes it ~work~.
stop_name = event["args"].replace(" ", "%20")
stop_search = Utils.get_url(URL_STOP_SEARCH % stop_name, get_params={
- "app_id": app_id, "app_key": app_key, "maxResults": "6", "faresOnly": "False"}, json=True)
+ "app_id": app_id, "app_key": app_key, "maxResults": "6",
+ "faresOnly": "False"}, json=True)
if stop_search:
for stop in stop_search["matches"]:
pass
- results = ["%s (%s): %s" % (stop["name"], ", ".join(stop["modes"]), stop["id"]) for stop in stop_search["matches"]]
- event["stdout"].write("[%s results] %s" % (stop_search["total"], "; ".join(results)))
+ results = ["%s (%s): %s" % (
+ stop["name"], ", ".join(stop["modes"]), stop["id"]) for stop in
+ stop_search["matches"]]
+ event["stdout"].write(
+ "[%s results] %s" % (stop_search["total"], "; ".join(results)))
else:
event["stderr"].write("No results")
@@ -207,14 +236,18 @@ class Module(object):
vehicle_id = event["args_split"][0]
vehicle = Utils.get_url(URL_VEHICLE % vehicle_id, get_params={
- "app_id": app_id, "app_key": app_key}, json=True)[0]
-
- arrival_time = self.vehicle_span(vehicle["expectedArrival"], human=False)
+ "app_id": app_id, "app_key": app_key}, json=True)[0]
+
+ arrival_time = self.vehicle_span(vehicle["expectedArrival"],
+ human=False)
platform = self.platform(vehicle["platformName"])
- event["stdout"].write("%s (%s) to %s. %s. Arrival at %s (%s) in %s minutes on %s" % (
- vehicle["vehicleId"], vehicle["lineName"], vehicle["destinationName"], vehicle["currentLocation"],
- vehicle["stationName"], vehicle["naptanId"], arrival_time, platform))
+ event["stdout"].write(
+ "%s (%s) to %s. %s. Arrival at %s (%s) in %s minutes on %s" % (
+ vehicle["vehicleId"], vehicle["lineName"],
+ vehicle["destinationName"], vehicle["currentLocation"],
+ vehicle["stationName"], vehicle["naptanId"], arrival_time,
+ platform))
def service(self, event):
app_id = self.bot.config["tfl-api-id"]
@@ -228,21 +261,27 @@ class Module(object):
return
results = self.result_map[event["target"].id]
if int(service_id) >= len(results):
- event["stdout"].write("%s is too high. Remember that the first arrival is 0" % service_id)
+ event["stdout"].write(
+ "%s is too high. Remember that the first arrival is 0" % service_id)
return
service = results[int(service_id)]
- arrivals = Utils.get_url(URL_LINE_ARRIVALS % service["route"], get_params={
- "app_id": app_id, "app_key": app_key}, json=True)
+ arrivals = Utils.get_url(URL_LINE_ARRIVALS % service["route"],
+ get_params={
+ "app_id": app_id, "app_key": app_key},
+ json=True)
arrivals = [a for a in arrivals if a["vehicleId"] == service["id"]]
arrivals = sorted(arrivals, key=lambda b: b["timeToStation"])
event["stdout"].write(
- "%s (%s) to %s: " % (arrivals[0]["vehicleId"], arrivals[0]["lineName"], arrivals[0]["destinationName"]) +
+ "%s (%s) to %s: " % (
+ arrivals[0]["vehicleId"], arrivals[0]["lineName"],
+ arrivals[0]["destinationName"]) +
", ".join(["%s (%s, %s)" %
- (a["stationName"], self.platform(a.get("platformName", "?"), True),
- a["expectedArrival"][11:16]
- ) for a in arrivals]))
+ (a["stationName"],
+ self.platform(a.get("platformName", "?"), True),
+ a["expectedArrival"][11:16]
+ ) for a in arrivals]))
def stop(self, event):
app_id = self.bot.config["tfl-api-id"]