Almtesh
/
infotbm
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Pull request from mdeboute 

Just a big refactoring
This commit is contained in:
Sasha MOREL 2022-05-08 13:57:34 +02:00 committed by GitHub
parent dd82e9afe9 5437def293
commit 323cbfbdc4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 503 additions and 409 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

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

12
README.md
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.

30
libs.py
View File

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

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())
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())
)

23
src/route.py Normal file
View File

@ -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 + ")"

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["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"
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)

28
src/utils.py Normal file
View File

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

114
src/vcub.py Normal file
View File

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

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)

126
vcub.py
View File

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