From aed901c0c0a943f3c87c144b97a02d449caf7574 Mon Sep 17 00:00:00 2001 From: mdeboute Date: Sat, 30 Apr 2022 01:43:20 +0200 Subject: [PATCH] refactor the lib --- README.md | 2 + src/main.py | 46 ++-- src/tbm_api/route.py | 23 ++ src/tbm_api/stop.py | 104 ++++++++ src/tbm_api/stop_area.py | 37 +++ src/tbm_api/stop_point.py | 26 ++ src/tbm_api/stop_route.py | 89 +++++++ src/{tbm_api_lib/libs.py => tbm_api/utils.py} | 4 - src/{tbm_api_lib => tbm_api}/vcub.py | 2 +- src/tbm_api/vehicle.py | 32 +++ src/tbm_api_lib/stop.py | 245 ------------------ 11 files changed, 337 insertions(+), 273 deletions(-) create mode 100644 src/tbm_api/route.py create mode 100644 src/tbm_api/stop.py create mode 100644 src/tbm_api/stop_area.py create mode 100644 src/tbm_api/stop_point.py create mode 100644 src/tbm_api/stop_route.py rename src/{tbm_api_lib/libs.py => tbm_api/utils.py} (96%) rename src/{tbm_api_lib => tbm_api}/vcub.py (98%) create mode 100644 src/tbm_api/vehicle.py delete mode 100644 src/tbm_api_lib/stop.py diff --git a/README.md b/README.md index 532292b..4166caf 100644 --- a/README.md +++ b/README.md @@ -1 +1,3 @@ # info_tbm + +A twitter bot to know if my tram will arrive soon ⏰ diff --git a/src/main.py b/src/main.py index f83eb6d..6a3969a 100644 --- a/src/main.py +++ b/src/main.py @@ -1,43 +1,43 @@ -from tbm_api_lib.stop import * import sys +from tbm_api.stop_area import * +from tbm_api.stop import * +from tbm_api.stop_route import StopRoute from datetime import datetime -from gmplot import gmplot + if __name__ == "__main__": for word in sys.argv[1:]: - for area in search_stop_by_name(word): - for stop in show_stops_from_ref(area["ref"])["stop_points"]: - for route in stop["routes"]: - if "Tram" in route["line_human"]: - sr = StopRoute(stop["id"], route["line_id"]) - line = sr.get_line() - for vehicule in line.vehicles(): + for area in get_stop_areas_by_name(word): + stop = get_stop_by_id(area.getId()) + for stopPoint in stop.getStopPoints(): + for route in stopPoint.getRoutes(): + if "Tram" in route.getLineName(): + stopRoute = StopRoute(stopPoint.getId(), route.getId()) + line = stopRoute.get_line() + for vehicule in line.get_vehicles(): v = line.get_vehicle(vehicule) - if v.is_realtime: + if v.getRealtime(): print( - str(v.wait_time_text) + str(v.getWaitTimeText()) + " (" - + datetime.fromtimestamp(v.arrival).strftime( + + datetime.fromtimestamp(v.getArrival()).strftime( "%H:%M" ) + ") → " - + v.destination - + " " - + str(v.location) + + v.getDestination() + + ", Curr location: " + + str(v.getLocation()) ) else: print( "~" - + str(v.wait_time_text) + + str(v.getWaitTimeText()) + " (" - + datetime.fromtimestamp(v.arrival).strftime( + + datetime.fromtimestamp(v.getArrival()).strftime( "%H:%M" ) + ") → " - + v.destination - + " " - + str(v.location) + + v.getDestination() + + ", Curr location: " + + str(v.getLocation()) ) - # gmap = gmplot.GoogleMapPlotter(v.location[0], v.location[1], 13) - # gmap.marker(v.location[0], v.location[1], "cornflowerblue") - # gmap.draw("map.html") diff --git a/src/tbm_api/route.py b/src/tbm_api/route.py new file mode 100644 index 0000000..ccb8a51 --- /dev/null +++ b/src/tbm_api/route.py @@ -0,0 +1,23 @@ +class Route: + def __init__(self, name, line_name): + self.id = None + self.name = name + self.line_name = line_name + + def getId(self): + return self.id + + def getName(self): + return self.name + + def getLineName(self): + return self.line_name + + def setId(self, id): + self.id = id + + def __repr__(self): + return self.name + " (" + self.line_name + ")" + + def __str__(self): + return self.name + " (" + self.line_name + ")" \ No newline at end of file diff --git a/src/tbm_api/stop.py b/src/tbm_api/stop.py new file mode 100644 index 0000000..16102e1 --- /dev/null +++ b/src/tbm_api/stop.py @@ -0,0 +1,104 @@ +from tbm_api.utils import get_data_from_json +from urllib.parse import quote +from tbm_api.stop_point import StopPoint +from re import search +from tbm_api.route import Route + + +INFO_URL = "https://ws.infotbm.com/ws/1.0/network/stoparea-informations/%s" + +LINE_TRANSLATE = { + "Tram A": "A", + "Tram B": "B", + "Tram C": "C", + "Tram D": "D", + "TBNight": "58", + "BAT3": "69", +} + +LINE_TYPES = ( + "Tram", + "Corol", + "Lianes", + "Ligne", + "Bus Relais", + "Citéis", +) + + +class Stop: + def __init__(self, id, name, latitude, longitude, city): + self.id = id + self.name = name + self.latitude = latitude + self.longitude = longitude + self.city = city + self.stopPoints = [] + + def getId(self): + return self.id + + def getName(self): + return self.name + + def getLatitude(self): + return self.latitude + + def getLongitude(self): + + return self.longitude + + def getCity(self): + return self.city + + def getStopPoints(self): + return self.stopPoints + + def setStopPoints(self, stopPoints): + self.stopPoints = stopPoints + + def __repr__(self): + return self.name + " (" + self.city + ")" + " (id: " + self.id + ")" + + def __str__(self): + return self.name + " (" + self.city + ")" + " (id: " + self.id + ")" + + +def get_stop_by_id(id): + data = get_data_from_json(INFO_URL % quote(id)) + stop = Stop( + data["id"], + data["name"], + float(data["latitude"]), + float(data["longitude"]), + data["city"], + ) + stopPoints = [] + for i in data["stopPoints"]: + stopPoint = StopPoint(i["name"]) + routes = [] + stopPoint.setId(int(search("[0-9]+$", i["id"]).group())) + for j in i["routes"]: + route = Route(j["name"], j["line"]["name"]) + add = False + if route.getLineName() in LINE_TRANSLATE: + line_id = LINE_TRANSLATE[route.getLineName()] + add = True + else: + try: + line_id = search("[0-9]+$", route.getLineName()).group() + except AttributeError: + continue + line_id = "%02d" % int(line_id) + for i in LINE_TYPES: + if route.getLineName()[0 : len(i)] == i: + add = True + break + if add: + route.setId(line_id) + routes.append(route) + stopPoint.setRoutes(routes) + if stopPoint.getRoutes() != []: + stopPoints.append(stopPoint) + stop.setStopPoints(stopPoints) + return stop diff --git a/src/tbm_api/stop_area.py b/src/tbm_api/stop_area.py new file mode 100644 index 0000000..b1f3479 --- /dev/null +++ b/src/tbm_api/stop_area.py @@ -0,0 +1,37 @@ +from tbm_api.utils import get_data_from_json +from urllib.parse import quote + + +STOP_URL = "https://ws.infotbm.com/ws/1.0/get-schedule/%s" + + +class StopArea: + def __init__(self, id, name, city): + self.id = id + self.name = name + self.city = city + + def getId(self): + return self.id + + def getName(self): + return self.name + + def getCity(self): + return self.city + + def __repr__(self): + return self.name + " (" + self.city + ")" + " (id: " + self.id + ")" + + def __str__(self): + return self.name + " (" + self.city + ")" + " (id: " + self.id + ")" + + +# we on only treat stops of type "stop_area" +def get_stop_areas_by_name(keyword): + data = get_data_from_json(STOP_URL % quote(keyword)) + stopAreas = [] + for s in data: + if s["type"] == "stop_area": + stopAreas.append(StopArea(s["id"], s["name"], s["city"])) + return stopAreas \ No newline at end of file diff --git a/src/tbm_api/stop_point.py b/src/tbm_api/stop_point.py new file mode 100644 index 0000000..903df09 --- /dev/null +++ b/src/tbm_api/stop_point.py @@ -0,0 +1,26 @@ +class StopPoint: + def __init__(self, name): + self.id = None + self.name = name + self.routes = [] + + def getId(self): + return self.id + + def getName(self): + return self.name + + def getRoutes(self): + return self.routes + + def setId(self, id): + self.id = id + + def setRoutes(self, routes): + self.routes = routes + + def __repr__(self): + return self.name + + def __str__(self): + return self.name \ No newline at end of file diff --git a/src/tbm_api/stop_route.py b/src/tbm_api/stop_route.py new file mode 100644 index 0000000..16f4a65 --- /dev/null +++ b/src/tbm_api/stop_route.py @@ -0,0 +1,89 @@ +from tbm_api.utils import get_data_from_json, hms2seconds +from time import time +from tbm_api.vehicle import Vehicle + + +SCHEDULE_URL = "https://ws.infotbm.com/ws/1.0/get-realtime-pass/%d/%s" + + +class Line: + """ + Information on the line served at a stop + """ + + def __init__(self, vehicles): + self.vehicles = vehicles + + def get_vehicles(self): + if self.vehicles is not None: + return list(range(0, len(self.vehicles))) + return [] + + def get_vehicle(self, vehicle_id): + return self.vehicles[vehicle_id] + + +class StopRoute: + def __init__( + self, + number, + line, + auto_update_at_creation=True, + auto_update=False, + auto_update_delay=-1, + ): + self.number = number + self.line = line + self.last_update = 0 + self.data = None + if auto_update_at_creation: + self.update() + + def update(self, auto=False): + """ + Update data + """ + + data = get_data_from_json(SCHEDULE_URL % (self.number, self.line)) + if "destinations" in data: + data = data["destinations"] + else: + print("No data for stop %d" % self.number) + return + self.last_update = time() + if type(data) == dict: + self.data = [] + # let's simplify the data + for i in data: + for j in data[i]: + location = None + try: + location = ( + float(j["vehicle_lattitude"]), + float(j["vehicle_longitude"]), + ) + except TypeError: + pass + vehicle = Vehicle( + j["vehicle_id"], + j["destination_name"], + j["realtime"] == "1", + location, + hms2seconds(j["waittime"]), + j["waittime_text"], + int(self.last_update + hms2seconds(j["waittime"])), + ) + self.data.append(vehicle) + self.data = sorted(self.data, key=lambda vehicle: vehicle.getArrival()) + else: + self.last_update = 0 + + def data_age(self): + """ + Returns the age of the data + """ + + return time() - self.last_update + + def get_line(self): + return Line(self.data) diff --git a/src/tbm_api_lib/libs.py b/src/tbm_api/utils.py similarity index 96% rename from src/tbm_api_lib/libs.py rename to src/tbm_api/utils.py index 9f2da9f..94129c9 100644 --- a/src/tbm_api_lib/libs.py +++ b/src/tbm_api/utils.py @@ -1,7 +1,3 @@ -""" -Common libraries -""" - from json import loads as read_json from urllib import request from urllib.error import HTTPError diff --git a/src/tbm_api_lib/vcub.py b/src/tbm_api/vcub.py similarity index 98% rename from src/tbm_api_lib/vcub.py rename to src/tbm_api/vcub.py index 0480eba..855772a 100644 --- a/src/tbm_api_lib/vcub.py +++ b/src/tbm_api/vcub.py @@ -2,7 +2,7 @@ Provides all info about V³ stations """ -from tbm_api_lib.libs import get_data_from_json +from tbm_api.utils import get_data_from_json from time import time vcub_url = "https://ws.infotbm.com/ws/1.0/vcubs" diff --git a/src/tbm_api/vehicle.py b/src/tbm_api/vehicle.py new file mode 100644 index 0000000..02d5bbe --- /dev/null +++ b/src/tbm_api/vehicle.py @@ -0,0 +1,32 @@ +class Vehicle: + def __init__( + self, id, destination, realtime, location, wait_time, wait_time_text, arrival + ): + self.id = id + self.destination = destination + self.realtime = realtime + self.location = location + self.wait_time = wait_time + self.wait_time_text = wait_time_text + self.arrival = arrival + + def getId(self): + return self.id + + def getDestination(self): + return self.destination + + def getRealtime(self): + return self.realtime + + def getLocation(self): + return self.location + + def getWaitTime(self): + return self.wait_time + + def getWaitTimeText(self): + return self.wait_time_text + + def getArrival(self): + return self.arrival diff --git a/src/tbm_api_lib/stop.py b/src/tbm_api_lib/stop.py deleted file mode 100644 index 0818e92..0000000 --- a/src/tbm_api_lib/stop.py +++ /dev/null @@ -1,245 +0,0 @@ -""" -Provides stop information -""" - -from tbm_api_lib.libs import get_data_from_json, hms2seconds -from time import time -from urllib.parse import quote -from re import search - - -STOP_URL = "https://ws.infotbm.com/ws/1.0/get-schedule/%s" -INFO_URL = "https://ws.infotbm.com/ws/1.0/network/stoparea-informations/%s" -SCHEDULE_URL = "https://ws.infotbm.com/ws/1.0/get-realtime-pass/%d/%s" - -line_types = ( - "Tram", - "Corol", - "Lianes", - "Ligne", - "Bus Relais", - "Citéis", -) -line_translate = { - "Tram A": "A", - "Tram B": "B", - "Tram C": "C", - "Tram D": "D", - "TBNight": "58", - "BAT3": "69", -} - - -def search_stop_by_name(keyword): - """ - Finds the reference of a stop name - - Format of data returned by the site - [ - { - id: str, named ref thereafter - name: str - type: str, but we only manage "stop_area" - city: str - }, - ] - """ - - d = get_data_from_json(STOP_URL % quote(keyword)) - r = [] - for i in d: - if i["type"] == "stop_area": - r.append( - { - "name": i["name"], - "city": i["city"], - "ref": i["id"], - } - ) - return r - - -def show_stops_from_ref(ref): - """ - Displays the list of stops of a reference given by search_stop_name - - Format of data returned by the site - { - id: str, content of given ref variable - name: str - latitude: str, convertible in float - longitude: str, convertible in float - city: str - hasWheelchairBoarding: bool, wheelchair accessibility - stopPoints: [ - id: str - name: str, once again the name of the stop - routes: [ - { - id: str - name: str, name of the terminus - line: { - name: str, human readable - } - } - ] - ] - } - """ - - d = get_data_from_json(INFO_URL % quote(ref)) - r = { - "ref": d["id"], - "name": d["name"], - "latitude": float(d["latitude"]), - "longitude": float(d["longitude"]), - "city": d["city"], - "stop_points": [], - } - for i in d["stopPoints"]: - s = { - "name": i["name"], - "routes": [], - } - s["id"] = int(search("[0-9]+$", i["id"]).group()) - for j in i["routes"]: - rte = { - "terminus": j["name"], - "line_human": j["line"]["name"], - } - add = False - if rte["line_human"] in line_translate: - line_id = line_translate[rte["line_human"]] - add = True - else: - try: - line_id = search("[0-9]+$", rte["line_human"]).group() - except AttributeError: - continue - line_id = "%02d" % int(line_id) - for i in line_types: - if rte["line_human"][0 : len(i)] == i: - add = True - break - if add: - rte["line_id"] = line_id - s["routes"].append(rte) - if s["routes"] != []: - r["stop_points"].append(s) - return r - - -class StopRoute: - """ - Retrieves information about a stop - - Format of data returned by the site - { - destinations: { - : [ - { - destination_name: str - realtime: 1 if followed, 0 otherwise - vehicle_id: str - vehicle_lattitude: float - vehicle_longitude: float - waittime: HH:MM:SS - waittime_text: str, human readable - }, - ] - } - } - """ - - def __init__( - self, - number, - line, - autoupdate_at_creation=True, - autoupdate=False, - autoupdate_delay=-1, - ): - self.number = number - self.line = line - self.last_update = 0 - self.data = None - if autoupdate_at_creation: - self.update() - - def update(self, auto=False): - """ - Update data - """ - - d = get_data_from_json(SCHEDULE_URL % (self.number, self.line)) - if "destinations" in d: - d = d["destinations"] - else: - return () - self.last_update = time() - if type(d) == dict: - self.data = [] - # let's simplify the data - for i in d: - for j in d[i]: - loc = None - try: - loc = ( - float(j["vehicle_lattitude"]), - float(j["vehicle_longitude"]), - ) - except TypeError: - pass - vehicle = { - "id": j["vehicle_id"], - "destination": j["destination_name"], - "realtime": j["realtime"] == "1", - "location": loc, - "wait_time": hms2seconds(j["waittime"]), - "wait_time_human": j["waittime_text"], - "arrival": int(self.last_update + hms2seconds(j["waittime"])), - } - self.data.append(vehicle) - self.data = sorted(self.data, key=lambda item: item["arrival"]) - else: - self.last_update = 0 - - def data_age(self): - """ - Returns the age of the data - """ - - return time() - self.last_update - - def get_line(self): - class Line: - """ - Information on the line served at a stop - """ - - def __init__(self, data): - self.ve = data - - def vehicles(self): - if self.ve is not None: - return list(range(0, len(self.ve))) - return [] - - def get_vehicle(self, vehicle): - class Vehicle: - """ - Information on a vehicle passage - """ - - def __init__(self, data): - self.id = data["id"] - self.location = data["location"] - self.destination = data["destination"] - self.is_realtime = data["realtime"] - self.wait_time = data["wait_time"] - self.wait_time_text = data["wait_time_human"] - self.arrival = data["arrival"] - - return Vehicle(self.ve[vehicle]) - - return Line(self.data)