Compare commits
8 Commits
dd82e9afe9
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
| e81da464e5 | |||
| c2d3dab3ee | |||
|
|
323cbfbdc4 | ||
|
|
5437def293 | ||
|
|
aed901c0c0 | ||
|
|
b69be7c05e | ||
|
|
0eedc2a638 | ||
|
|
6a2c514265 |
10
README.md
10
README.md
@@ -1,13 +1,13 @@
|
|||||||
# What's this?
|
# 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
43
src/main.py
Normal 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
27
src/route.py
Normal 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
104
src/stop.py
Normal 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
37
src/stop_area.py
Normal 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
26
src/stop_point.py
Normal 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
89
src/stop_route.py
Normal 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)
|
||||||
@@ -1,7 +1,3 @@
|
|||||||
'''
|
|
||||||
Common libraries
|
|
||||||
'''
|
|
||||||
|
|
||||||
from json import loads as read_json
|
from json import loads as read_json
|
||||||
from urllib import request
|
from urllib import request
|
||||||
from urllib.error import HTTPError
|
from urllib.error import HTTPError
|
||||||
@@ -9,22 +5,24 @@ from urllib.error import HTTPError
|
|||||||
|
|
||||||
def get_data_from_json (url):
|
def get_data_from_json (url):
|
||||||
'''
|
'''
|
||||||
gets data from json at url
|
Gets data from json at url.
|
||||||
'''
|
'''
|
||||||
|
|
||||||
opener = request.build_opener ()
|
opener = request.build_opener ()
|
||||||
try:
|
try:
|
||||||
return (read_json (opener.open (url).read ().decode ('utf8')))
|
return read_json (opener.open (url).read ().decode ('utf8'))
|
||||||
except HTTPError:
|
except HTTPError:
|
||||||
return (None)
|
return None
|
||||||
|
|
||||||
|
|
||||||
def hms2seconds (hhmmss):
|
def hms2seconds (hhmmss):
|
||||||
'''
|
'''
|
||||||
Convert H:M:S string to time in seconds
|
Convert H:M:S string to time in seconds.
|
||||||
'''
|
'''
|
||||||
|
|
||||||
try:
|
try:
|
||||||
cut_string = hhmmss.split (':')
|
cut_string = hhmmss.split (':')
|
||||||
cut_time = (int (cut_string [0]), int (cut_string [1]), int (cut_string [2]))
|
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):
|
except (IndexError, ValueError, TypeError):
|
||||||
return (0)
|
return None
|
||||||
113
src/vcub.py
Normal file
113
src/vcub.py
Normal 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
32
src/vehicle.py
Normal 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
246
stop.py
@@ -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
126
vcub.py
@@ -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 ())
|
|
||||||
Reference in New Issue
Block a user