diff options
| author | 2018-08-31 10:50:37 +0100 | |
|---|---|---|
| committer | 2018-08-31 10:50:37 +0100 | |
| commit | abed9cf4ea71dcbad2dd2c049683b8d14b942e09 (patch) | |
| tree | 3e40caf63fa7e1500469f4ad9a0c45c51808aad4 /modules/tfl.py | |
| parent | Fix 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.py | 171 |
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"] |
