Compare commits

..

11 Commits

Author SHA1 Message Date
e81da464e5 Modification de l'URL des infos de passage
L'identifiant de ligne est maintenant requis à la fin de l'URL.
Cette information n'étant pas utile auparavent, j'ai donc ajouté sa récupération et son stockage.
2023-08-26 11:33:49 +02:00
c2d3dab3ee Affichage dans la méthode StopRoute.update ()
La méthode StopRoute.update () doit juste mettre à jour les données.
Si ce n'est pas possible, l'erreur doit être retournée d'un façon exploitable par le code qui y fait appel.
Cette correction supprime seulement l'affichage dans la sortie standard, le code qui y fait appel n'aura l'information de l'échec de la mise à jour des données qu'en utilisant la fonction StopRoute.data_age ().
2023-05-01 08:00:41 +02:00
Sasha MOREL
323cbfbdc4 Pull request from mdeboute
Just a big refactoring
2022-05-08 13:57:34 +02:00
mdeboute
5437def293 clarify things 2022-05-08 13:17:07 +02:00
mdeboute
aed901c0c0 refactor the lib 2022-04-30 01:43:20 +02:00
mdeboute
b69be7c05e cool! 2022-04-28 12:03:51 +02:00
mdeboute
0eedc2a638 k 2022-04-26 22:32:25 +02:00
mdeboute
6a2c514265 k 2022-04-26 22:25:40 +02:00
dd82e9afe9 Listage des ids de toutes les stations 2021-11-13 10:14:51 +01:00
c80eeac919 Erreur de structure de données
Je modifie la structure de données au chargement pour une utilisation plus facile et rapide, il faut gérer les données brutes.
2021-11-13 10:02:02 +01:00
87283d36e0 V³ : ajout de l'utilisation de données déjà disponibles 2021-11-13 09:36:02 +01:00
13 changed files with 486 additions and 376 deletions

2
.gitignore vendored
View File

@@ -1,3 +1,3 @@
# Byte-compiled Python
__pycache__/
*.pyc
*.pyc

View File

@@ -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.
These libraries are under GNU GPL v3. See [license](LICENSE) for more details.

43
src/main.py Normal file
View File

@@ -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 (), route.getLineId ())
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 ())
)

27
src/route.py Normal file
View File

@@ -0,0 +1,27 @@
class Route:
def __init__ (self, line_id, name, line_name):
self.id = None
self.name = name
self.line_name = line_name
self.line_id = line_id
def getId (self):
return self.id
def getName (self):
return self.name
def getLineName (self):
return self.line_name
def getLineId (self):
return self.line_id
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 + ')'

104
src/stop.py Normal file
View File

@@ -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 ['id'], 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

37
src/stop_area.py Normal file
View File

@@ -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

26
src/stop_point.py Normal file
View File

@@ -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

89
src/stop_route.py Normal file
View File

@@ -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/%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,
line_id,
auto_update_at_creation = True,
auto_update = False,
auto_update_delay = -1,
):
self.number = number
self.line = line
self.line_id = line_id
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, self.line_id))
if 'destinations' in data:
data = data ['destinations']
else:
return
self.last_update = time ()
if type (data) == dict:
self.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)

View File

@@ -1,7 +1,3 @@
'''
Common libraries
'''
from json import loads as read_json
from urllib import request
from urllib.error import HTTPError
@@ -9,22 +5,24 @@ from urllib.error import HTTPError
def get_data_from_json (url):
'''
gets data from json at url
Gets data from json at url.
'''
opener = request.build_opener ()
try:
return (read_json (opener.open (url).read ().decode ('utf8')))
return read_json (opener.open (url).read ().decode ('utf8'))
except HTTPError:
return (None)
return None
def hms2seconds (hhmmss):
'''
Convert H:M:S string to time in seconds
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])
return 3600 * cut_time [0] + 60 * cut_time [1] + cut_time [2]
except (IndexError, ValueError, TypeError):
return (0)
return None

113
src/vcub.py Normal file
View File

@@ -0,0 +1,113 @@
"""
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
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 ())

32
src/vehicle.py Normal file
View File

@@ -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

246
stop.py
View File

@@ -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_stop_id>: [
{
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)

113
vcub.py
View File

@@ -1,113 +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 = True, autoupdate = False, autoupdate_delay = -1):
self.last_update = 0
self.data = None
if autoupdate_at_creation:
self.update ()
def update (self, auto = False):
'''
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
'''
d = get_data_from_json (vcub_url)
# 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))
if __name__ == '__main__':
v = Vcub ()
for i in (v.get_by_id (149), v.get_by_id (v.get_names () ['Buttiniere']), 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))