diff --git a/.gitignore b/.gitignore index e156e60..204377b 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,3 @@ # Byte-compiled Python __pycache__/ -*.pyc \ No newline at end of file +*.pyc diff --git a/README.md b/README.md index bf652f5..f71abec 100644 --- a/README.md +++ b/README.md @@ -1,13 +1,13 @@ # What's this? -This is a set of libraries to gather data from TBM. +This is a set of libraries to gather real time data from TBM. -They all work around [the dynamic map](http://plandynamique.infotbm.com). +They all work around [the dynamic map](https://www.infotbm.com/fr/plans/plan-dynamique/). -# What's TBM? +## What's TBM? -TBM is a french compagny that operates Bordeaux's public transportation. +TBM is a French public transport company in Bordeaux. -# Licence +## Licence -These libraries are under GNU GPL v3. See [license](LICENSE) for more details. \ No newline at end of file +These libraries are under GNU GPL v3. See [license](LICENSE) for more details. diff --git a/libs.py b/libs.py deleted file mode 100644 index e823917..0000000 --- a/libs.py +++ /dev/null @@ -1,30 +0,0 @@ -''' -Common libraries -''' - -from json import loads as read_json -from urllib import request -from urllib.error import HTTPError - - -def get_data_from_json (url): - ''' - gets data from json at url - ''' - opener = request.build_opener () - try: - return (read_json (opener.open (url).read ().decode ('utf8'))) - except HTTPError: - return (None) - - -def hms2seconds (hhmmss): - ''' - Convert H:M:S string to time in seconds - ''' - try: - cut_string = hhmmss.split (':') - cut_time = (int (cut_string [0]), int (cut_string [1]), int (cut_string [2])) - return (3600 * cut_time [0] + 60 * cut_time [1] + cut_time [2]) - except (IndexError, ValueError, TypeError): - return (0) \ No newline at end of file diff --git a/src/main.py b/src/main.py new file mode 100644 index 0000000..e03f4c4 --- /dev/null +++ b/src/main.py @@ -0,0 +1,43 @@ +import sys +from stop_area import * +from stop import * +from stop_route import StopRoute +from datetime import datetime + + +if __name__ == "__main__": + for word in sys.argv[1:]: + 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.getRealtime(): + print( + str(v.getWaitTimeText()) + + " (" + + datetime.fromtimestamp(v.getArrival()).strftime( + "%H:%M" + ) + + ") → " + + v.getDestination() + + ", Curr location: " + + str(v.getLocation()) + ) + else: + print( + "~" + + str(v.getWaitTimeText()) + + " (" + + datetime.fromtimestamp(v.getArrival()).strftime( + "%H:%M" + ) + + ") → " + + v.getDestination() + + ", Curr location: " + + str(v.getLocation()) + ) diff --git a/src/route.py b/src/route.py new file mode 100644 index 0000000..ccb8a51 --- /dev/null +++ b/src/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/stop.py b/src/stop.py new file mode 100644 index 0000000..5d96a19 --- /dev/null +++ b/src/stop.py @@ -0,0 +1,104 @@ +from utils import get_data_from_json +from urllib.parse import quote +from stop_point import StopPoint +from re import search +from 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/stop_area.py b/src/stop_area.py new file mode 100644 index 0000000..3dbeb3e --- /dev/null +++ b/src/stop_area.py @@ -0,0 +1,37 @@ +from 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/stop_point.py b/src/stop_point.py new file mode 100644 index 0000000..903df09 --- /dev/null +++ b/src/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/stop_route.py b/src/stop_route.py new file mode 100644 index 0000000..c769a31 --- /dev/null +++ b/src/stop_route.py @@ -0,0 +1,89 @@ +from utils import get_data_from_json, hms2seconds +from time import time +from 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/utils.py b/src/utils.py new file mode 100644 index 0000000..90314d2 --- /dev/null +++ b/src/utils.py @@ -0,0 +1,28 @@ +from json import loads as read_json +from urllib import request +from urllib.error import HTTPError + + +def get_data_from_json(url): + """ + Gets data from json at url. + """ + + opener = request.build_opener() + try: + return read_json(opener.open(url).read().decode("utf8")) + except HTTPError: + return None + + +def hms2seconds(hhmmss): + """ + Convert H:M:S string to time in seconds. + """ + + try: + cut_string = hhmmss.split(":") + cut_time = (int(cut_string[0]), int(cut_string[1]), int(cut_string[2])) + return 3600 * cut_time[0] + 60 * cut_time[1] + cut_time[2] + except (IndexError, ValueError, TypeError): + return None diff --git a/src/vcub.py b/src/vcub.py new file mode 100644 index 0000000..259bddd --- /dev/null +++ b/src/vcub.py @@ -0,0 +1,114 @@ +""" +Provides all info about V³ stations. +""" + +from utils import get_data_from_json +from time import time + + +vcub_url = "https://ws.infotbm.com/ws/1.0/vcubs" + + +class Vcub: + """ + Retrieves information from V³ stations. + """ + + def __init__( + self, + autoupdate_at_creation=None, + autoupdate=False, + autoupdate_delay=-1, + data=None, + ): + self.last_update = 0 + if type(data) == dict: + self.data = self.update(data=data) + else: + self.data = None + if autoupdate_at_creation or ( + autoupdate_at_creation is None and self.data is None + ): + self.update() + + def update(self, auto=False, data=None): + """ + Updates data. + """ + + if data is None or type(data) != dict: + d = get_data_from_json(vcub_url) + else: + d = data + # the original format is awfull, so I change it a little + if type(d) == dict: + self.data = {} + d = d["lists"] + for i in d: + e = { + "name": i["name"], + "online": i["connexionState"] == "CONNECTEE", + "plus": i["typeVlsPlus"] == "VLS_PLUS", + "empty": int(i["nbPlaceAvailable"]), + "bikes": int(i["nbBikeAvailable"]), + "ebikes": int(i["nbElectricBikeAvailable"]), + "location": (float(i["latitude"]), float(i["longitude"])), + } + self.data[int(i["id"])] = e + self.last_update = time() + + def data_age(self): + """ + Computes the data's age. + """ + + return time() - self.last_update + + def get_names(self): + """ + Returns all names in a dict with id as data. + """ + + r = {} + for i in self.data: + r[self.data[i]["name"]] = i + return r + + def get_locations(self): + """ + Returns all locations in a dict with id as data. + """ + + r = {} + for i in self.data: + r[self.data[i]["location"]] = i + return r + + def get_by_id(self, id): + """ + Returns a station by its id. + """ + + class Station: + """ + A V³ station + """ + + def __init__(self, data, id): + self.data = data + self.id = id + self.name = self.data["name"] + self.location = self.data["location"] + self.online = self.data["online"] + self.isplus = self.data["plus"] + self.bikes = self.data["bikes"] + self.ebikes = self.data["ebikes"] + self.empty = self.data["empty"] + + def __int__(self): + return self.id + + return Station(self.data[id], id) + + def get_all_ids(self): + return tuple(self.data.keys()) diff --git a/src/vehicle.py b/src/vehicle.py new file mode 100644 index 0000000..02d5bbe --- /dev/null +++ b/src/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/stop.py b/stop.py deleted file mode 100644 index c0b35cf..0000000 --- a/stop.py +++ /dev/null @@ -1,246 +0,0 @@ -''' -Fourni les informations sur les arrêts -''' - -from libs import get_data_from_json, hms2seconds -from time import time -from urllib.parse import quote -from re import search - -search_stop_url = 'https://ws.infotbm.com/ws/1.0/get-schedule/%s' -stop_info_url = 'https://ws.infotbm.com/ws/1.0/network/stoparea-informations/%s' -stop_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): - ''' - Recherche la référence d'un nom d'arrêt - - Format des données retournées par le site - [ - { - id: str, nommé ref par la suite - name: str - type: str, mais je ne gère que "stop_area" - city: str - }, - ] - ''' - d = get_data_from_json (search_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): - ''' - Affiche la liste des arrêts d'une référence donnée par search_stop_name - - Format des données retournées par le site - { - id: str, contenu de la variable ref donnée - name: str - latitude: str, convertible en float - longitude: str, convertible en float - city: str - hasWheelchairBoarding: bool, accessibilité en fauteuil roulant - stopPoints: [ - id: str - name: str, encore le nom - routes: [ - { - id: str - name: str, nom du terminus - line: { - name: str, nom pour les humains - } - } - ] - ] - } - ''' - d = get_data_from_json (stop_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 (): - ''' - Récupère les informations sur un arrêt - - Format des données retournées pas le site - { - destinations: { - : [ - { - destination_name: str - realtime: 1 si suivi, 0 sinon - vehicle_id: str - vehicle_lattitude: float - vehicle_longitude: float - waittime: HH:MM:SS - waittime_text: str, lisible pas un humain - }, - ] - } - } - ''' - - 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): - ''' - Met à jour les données - ''' - d = get_data_from_json (stop_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): - ''' - Retourne l'âge des données - ''' - return (time () - self.last_update) - - def get_line (self): - class Line (): - ''' - Information sur la ligne déservie à un arrêt - ''' - 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 sur un passage de véhicule - ''' - 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)) - - -if __name__ == '__main__': - from datetime import datetime - for word in ('Gravière', 'Gare Saint Jean', 'Quinconces', 'Zorbut'): - print (word + ':') - for area in search_stop_by_name (word): - print ('\t' + area ['name'] + ' (' + area ['city'] + '):') - for stop in show_stops_from_ref (area ['ref']) ['stop_points']: - print ('\t\t' + stop ['name'] + ' (' + str (stop ['id']) + '):') - for route in stop ['routes']: - print ('\t\t\t' + route ['line_human'] + ' terminus ' + route ['terminus'] + ':') - sr = StopRoute (stop ['id'], route ['line_id']) - line = sr.get_line () - for vehicle in line.vehicles (): - v = line.get_vehicle (vehicle) - if v.is_realtime: - print ('\t\t\t\t' + str (v.wait_time) + ' (' + datetime.fromtimestamp (v.arrival).strftime ('%H:%M') + ') → ' + v.destination) - else: - print ('\t\t\t\t~' + str (v.wait_time) + ' (' + datetime.fromtimestamp (v.arrival).strftime ('%H:%M') + ') → ' + v.destination) \ No newline at end of file diff --git a/vcub.py b/vcub.py deleted file mode 100644 index 9d814e9..0000000 --- a/vcub.py +++ /dev/null @@ -1,126 +0,0 @@ -''' -Provides all info about V³ stations -''' - -from libs import get_data_from_json -from time import time - -vcub_url = 'https://ws.infotbm.com/ws/1.0/vcubs' - - -class Vcub (): - ''' - Récupère les informations des stations V³ - - Format de données, tel que retourné par le site infotbm : - { - lists: [ - { - 'id': numéro de la station, - 'name': str, - 'connexionState': 'CONNECTEE' si en service 'DECONNECTEE' sinon, - 'typeVlsPlus': 'VLS_PLUS' si V³+ 'PAS_VLS_PLUS' sinon, - 'nbPlaceAvailable': 33, - 'nbBikeAvailable': 0 - 'nbElectricBikeAvailable': 6 - 'latitude': float, - 'longitude': float, - }, - ] - } - ''' - def __init__ (self, autoupdate_at_creation = None, autoupdate = False, autoupdate_delay = -1, data = None): - self.last_update = 0 - if type (data) == dict: - self.data = self.update (data = data) - else: - self.data = None - if autoupdate_at_creation or (autoupdate_at_creation is None and self.data is None): - self.update () - - def update (self, auto = False, data = None): - ''' - Updates data - auto optionnal param is to set if a update is a automatic one, and must be performed as defined in autoupdate_delay variable - ''' - if data is None or type (data) != dict: - d = get_data_from_json (vcub_url) - else: - d = data - # the original format is awfull, so I change it a little - if type (d) == dict: - self.data = {} - d = d ['lists'] - for i in d: - e = { - 'name': i ['name'], - 'online': i ['connexionState'] == 'CONNECTEE', - 'plus': i ['typeVlsPlus'] == 'VLS_PLUS', - 'empty': int (i ['nbPlaceAvailable']), - 'bikes': int (i ['nbBikeAvailable']), - 'ebikes': int (i ['nbElectricBikeAvailable']), - 'location': (float (i ['latitude']), float (i ['longitude'])) - } - self.data [int (i ['id'])] = e - self.last_update = time () - - def data_age (self): - ''' - Computes the data's age - ''' - return (time () - self.last_update) - - def get_names (self): - ''' - Returns all names in a dict with id as data - ''' - r = {} - for i in self.data: - r [self.data [i] ['name']] = i - return (r) - - def get_locations (self): - ''' - Returns all locations in a dict with id as data - ''' - r = {} - for i in self.data: - r [self.data [i] ['location']] = i - return (r) - - def get_by_id (self, id): - ''' - Returns a station by its id - ''' - class Station (): - ''' - A V³ station - ''' - def __init__ (self, data, id): - self.data = data - self.id = id - self.name = self.data ['name'] - self.location = self.data ['location'] - self.online = self.data ['online'] - self.isplus = self.data ['plus'] - self.bikes = self.data ['bikes'] - self.ebikes = self.data ['ebikes'] - self.empty = self.data ['empty'] - - def __int__ (self): - return (self.id) - - return (Station (self.data [id], id)) - - def get_all_ids (self): - return (tuple (self.data.keys ())) - - -if __name__ == '__main__': - v = Vcub () - for i in (v.get_by_id (149), v.get_by_id (v.get_names () ['Buttiniere']), ): - print ('%s (%d) (%f, %f)%s%s\n\tbikes: %d\n\te-bikes: %d\n\tfree: %d\n\t' % (i.name, i, i.location [0], i.location [1], i.isplus and ' (VCUB+)' or '', i.online and ' ' or ' OFFLINE', i.bikes, i.ebikes, i.empty)) - v = Vcub (data = get_data_from_json (vcub_url)) - for i in (v.get_by_id (v.get_locations () [(44.8875, -0.51763)]), ): - print ('%s (%d) (%f, %f)%s%s\n\tbikes: %d\n\te-bikes: %d\n\tfree: %d\n\t' % (i.name, i, i.location [0], i.location [1], i.isplus and ' (VCUB+)' or '', i.online and ' ' or ' OFFLINE', i.bikes, i.ebikes, i.empty)) - print ('stations :', v.get_all_ids ()) \ No newline at end of file