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

refactor the lib

This commit is contained in:
mdeboute 2022-04-30 01:43:20 +02:00
parent b69be7c05e
commit aed901c0c0
11 changed files with 337 additions and 273 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -1 +1,3 @@
# info_tbm
A twitter bot to know if my tram will arrive soon ⏰

46
src/main.py
View File

@ -1,43 +1,43 @@
from tbm_api_lib.stop import *
import sys
from tbm_api.stop_area import *
from tbm_api.stop import *
from tbm_api.stop_route import StopRoute
from datetime import datetime
from gmplot import gmplot
if __name__ == "__main__":
for word in sys.argv[1:]:
for area in search_stop_by_name(word):
for stop in show_stops_from_ref(area["ref"])["stop_points"]:
for route in stop["routes"]:
if "Tram" in route["line_human"]:
sr = StopRoute(stop["id"], route["line_id"])
line = sr.get_line()
for vehicule in line.vehicles():
for area in get_stop_areas_by_name(word):
stop = get_stop_by_id(area.getId())
for stopPoint in stop.getStopPoints():
for route in stopPoint.getRoutes():
if "Tram" in route.getLineName():
stopRoute = StopRoute(stopPoint.getId(), route.getId())
line = stopRoute.get_line()
for vehicule in line.get_vehicles():
v = line.get_vehicle(vehicule)
if v.is_realtime:
if v.getRealtime():
print(
str(v.wait_time_text)
str(v.getWaitTimeText())
+ " ("
+ datetime.fromtimestamp(v.arrival).strftime(
+ datetime.fromtimestamp(v.getArrival()).strftime(
"%H:%M"
)
+ ") → "
+ v.destination
+ " "
+ str(v.location)
+ v.getDestination()
+ ", Curr location: "
+ str(v.getLocation())
)
else:
print(
"~"
+ str(v.wait_time_text)
+ str(v.getWaitTimeText())
+ " ("
+ datetime.fromtimestamp(v.arrival).strftime(
+ datetime.fromtimestamp(v.getArrival()).strftime(
"%H:%M"
)
+ ") → "
+ v.destination
+ " "
+ str(v.location)
+ v.getDestination()
+ ", Curr location: "
+ str(v.getLocation())
)
# gmap = gmplot.GoogleMapPlotter(v.location[0], v.location[1], 13)
# gmap.marker(v.location[0], v.location[1], "cornflowerblue")
# gmap.draw("map.html")

23
src/tbm_api/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/tbm_api/stop.py Normal file
View File

@ -0,0 +1,104 @@
from tbm_api.utils import get_data_from_json
from urllib.parse import quote
from tbm_api.stop_point import StopPoint
from re import search
from tbm_api.route import Route
INFO_URL = "https://ws.infotbm.com/ws/1.0/network/stoparea-informations/%s"
LINE_TRANSLATE = {
"Tram A": "A",
"Tram B": "B",
"Tram C": "C",
"Tram D": "D",
"TBNight": "58",
"BAT3": "69",
}
LINE_TYPES = (
"Tram",
"Corol",
"Lianes",
"Ligne",
"Bus Relais",
"Citéis",
)
class Stop:
def __init__(self, id, name, latitude, longitude, city):
self.id = id
self.name = name
self.latitude = latitude
self.longitude = longitude
self.city = city
self.stopPoints = []
def getId(self):
return self.id
def getName(self):
return self.name
def getLatitude(self):
return self.latitude
def getLongitude(self):
return self.longitude
def getCity(self):
return self.city
def getStopPoints(self):
return self.stopPoints
def setStopPoints(self, stopPoints):
self.stopPoints = stopPoints
def __repr__(self):
return self.name + " (" + self.city + ")" + " (id: " + self.id + ")"
def __str__(self):
return self.name + " (" + self.city + ")" + " (id: " + self.id + ")"
def get_stop_by_id(id):
data = get_data_from_json(INFO_URL % quote(id))
stop = Stop(
data["id"],
data["name"],
float(data["latitude"]),
float(data["longitude"]),
data["city"],
)
stopPoints = []
for i in data["stopPoints"]:
stopPoint = StopPoint(i["name"])
routes = []
stopPoint.setId(int(search("[0-9]+$", i["id"]).group()))
for j in i["routes"]:
route = Route(j["name"], j["line"]["name"])
add = False
if route.getLineName() in LINE_TRANSLATE:
line_id = LINE_TRANSLATE[route.getLineName()]
add = True
else:
try:
line_id = search("[0-9]+$", route.getLineName()).group()
except AttributeError:
continue
line_id = "%02d" % int(line_id)
for i in LINE_TYPES:
if route.getLineName()[0 : len(i)] == i:
add = True
break
if add:
route.setId(line_id)
routes.append(route)
stopPoint.setRoutes(routes)
if stopPoint.getRoutes() != []:
stopPoints.append(stopPoint)
stop.setStopPoints(stopPoints)
return stop

37
src/tbm_api/stop_area.py Normal file
View File

@ -0,0 +1,37 @@
from tbm_api.utils import get_data_from_json
from urllib.parse import quote
STOP_URL = "https://ws.infotbm.com/ws/1.0/get-schedule/%s"
class StopArea:
def __init__(self, id, name, city):
self.id = id
self.name = name
self.city = city
def getId(self):
return self.id
def getName(self):
return self.name
def getCity(self):
return self.city
def __repr__(self):
return self.name + " (" + self.city + ")" + " (id: " + self.id + ")"
def __str__(self):
return self.name + " (" + self.city + ")" + " (id: " + self.id + ")"
# we on only treat stops of type "stop_area"
def get_stop_areas_by_name(keyword):
data = get_data_from_json(STOP_URL % quote(keyword))
stopAreas = []
for s in data:
if s["type"] == "stop_area":
stopAreas.append(StopArea(s["id"], s["name"], s["city"]))
return stopAreas

26
src/tbm_api/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/tbm_api/stop_route.py Normal file
View File

@ -0,0 +1,89 @@
from tbm_api.utils import get_data_from_json, hms2seconds
from time import time
from tbm_api.vehicle import Vehicle
SCHEDULE_URL = "https://ws.infotbm.com/ws/1.0/get-realtime-pass/%d/%s"
class Line:
"""
Information on the line served at a stop
"""
def __init__(self, vehicles):
self.vehicles = vehicles
def get_vehicles(self):
if self.vehicles is not None:
return list(range(0, len(self.vehicles)))
return []
def get_vehicle(self, vehicle_id):
return self.vehicles[vehicle_id]
class StopRoute:
def __init__(
self,
number,
line,
auto_update_at_creation=True,
auto_update=False,
auto_update_delay=-1,
):
self.number = number
self.line = line
self.last_update = 0
self.data = None
if auto_update_at_creation:
self.update()
def update(self, auto=False):
"""
Update data
"""
data = get_data_from_json(SCHEDULE_URL % (self.number, self.line))
if "destinations" in data:
data = data["destinations"]
else:
print("No data for stop %d" % self.number)
return
self.last_update = time()
if type(data) == dict:
self.data = []
# let's simplify the data
for i in data:
for j in data[i]:
location = None
try:
location = (
float(j["vehicle_lattitude"]),
float(j["vehicle_longitude"]),
)
except TypeError:
pass
vehicle = Vehicle(
j["vehicle_id"],
j["destination_name"],
j["realtime"] == "1",
location,
hms2seconds(j["waittime"]),
j["waittime_text"],
int(self.last_update + hms2seconds(j["waittime"])),
)
self.data.append(vehicle)
self.data = sorted(self.data, key=lambda vehicle: vehicle.getArrival())
else:
self.last_update = 0
def data_age(self):
"""
Returns the age of the data
"""
return time() - self.last_update
def get_line(self):
return Line(self.data)

4
src/tbm_api_lib/libs.py → src/tbm_api/utils.py
View File

@ -1,7 +1,3 @@
"""
Common libraries
"""
from json import loads as read_json
from urllib import request
from urllib.error import HTTPError

2
src/tbm_api_lib/vcub.py → src/tbm_api/vcub.py
View File

@ -2,7 +2,7 @@
Provides all info about V³ stations
"""
from tbm_api_lib.libs import get_data_from_json
from tbm_api.utils import get_data_from_json
from time import time
vcub_url = "https://ws.infotbm.com/ws/1.0/vcubs"

32
src/tbm_api/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

245
src/tbm_api_lib/stop.py
View File

@ -1,245 +0,0 @@
"""
Provides stop information
"""
from tbm_api_lib.libs import get_data_from_json, hms2seconds
from time import time
from urllib.parse import quote
from re import search
STOP_URL = "https://ws.infotbm.com/ws/1.0/get-schedule/%s"
INFO_URL = "https://ws.infotbm.com/ws/1.0/network/stoparea-informations/%s"
SCHEDULE_URL = "https://ws.infotbm.com/ws/1.0/get-realtime-pass/%d/%s"
line_types = (
"Tram",
"Corol",
"Lianes",
"Ligne",
"Bus Relais",
"Citéis",
)
line_translate = {
"Tram A": "A",
"Tram B": "B",
"Tram C": "C",
"Tram D": "D",
"TBNight": "58",
"BAT3": "69",
}
def search_stop_by_name(keyword):
"""
Finds the reference of a stop name
Format of data returned by the site
[
{
id: str, named ref thereafter
name: str
type: str, but we only manage "stop_area"
city: str
},
]
"""
d = get_data_from_json(STOP_URL % quote(keyword))
r = []
for i in d:
if i["type"] == "stop_area":
r.append(
{
"name": i["name"],
"city": i["city"],
"ref": i["id"],
}
)
return r
def show_stops_from_ref(ref):
"""
Displays the list of stops of a reference given by search_stop_name
Format of data returned by the site
{
id: str, content of given ref variable
name: str
latitude: str, convertible in float
longitude: str, convertible in float
city: str
hasWheelchairBoarding: bool, wheelchair accessibility
stopPoints: [
id: str
name: str, once again the name of the stop
routes: [
{
id: str
name: str, name of the terminus
line: {
name: str, human readable
}
}
]
]
}
"""
d = get_data_from_json(INFO_URL % quote(ref))
r = {
"ref": d["id"],
"name": d["name"],
"latitude": float(d["latitude"]),
"longitude": float(d["longitude"]),
"city": d["city"],
"stop_points": [],
}
for i in d["stopPoints"]:
s = {
"name": i["name"],
"routes": [],
}
s["id"] = int(search("[0-9]+$", i["id"]).group())
for j in i["routes"]:
rte = {
"terminus": j["name"],
"line_human": j["line"]["name"],
}
add = False
if rte["line_human"] in line_translate:
line_id = line_translate[rte["line_human"]]
add = True
else:
try:
line_id = search("[0-9]+$", rte["line_human"]).group()
except AttributeError:
continue
line_id = "%02d" % int(line_id)
for i in line_types:
if rte["line_human"][0 : len(i)] == i:
add = True
break
if add:
rte["line_id"] = line_id
s["routes"].append(rte)
if s["routes"] != []:
r["stop_points"].append(s)
return r
class StopRoute:
"""
Retrieves information about a stop
Format of data returned by the site
{
destinations: {
<destination_stop_id>: [
{
destination_name: str
realtime: 1 if followed, 0 otherwise
vehicle_id: str
vehicle_lattitude: float
vehicle_longitude: float
waittime: HH:MM:SS
waittime_text: str, human readable
},
]
}
}
"""
def __init__(
self,
number,
line,
autoupdate_at_creation=True,
autoupdate=False,
autoupdate_delay=-1,
):
self.number = number
self.line = line
self.last_update = 0
self.data = None
if autoupdate_at_creation:
self.update()
def update(self, auto=False):
"""
Update data
"""
d = get_data_from_json(SCHEDULE_URL % (self.number, self.line))
if "destinations" in d:
d = d["destinations"]
else:
return ()
self.last_update = time()
if type(d) == dict:
self.data = []
# let's simplify the data
for i in d:
for j in d[i]:
loc = None
try:
loc = (
float(j["vehicle_lattitude"]),
float(j["vehicle_longitude"]),
)
except TypeError:
pass
vehicle = {
"id": j["vehicle_id"],
"destination": j["destination_name"],
"realtime": j["realtime"] == "1",
"location": loc,
"wait_time": hms2seconds(j["waittime"]),
"wait_time_human": j["waittime_text"],
"arrival": int(self.last_update + hms2seconds(j["waittime"])),
}
self.data.append(vehicle)
self.data = sorted(self.data, key=lambda item: item["arrival"])
else:
self.last_update = 0
def data_age(self):
"""
Returns the age of the data
"""
return time() - self.last_update
def get_line(self):
class Line:
"""
Information on the line served at a stop
"""
def __init__(self, data):
self.ve = data
def vehicles(self):
if self.ve is not None:
return list(range(0, len(self.ve)))
return []
def get_vehicle(self, vehicle):
class Vehicle:
"""
Information on a vehicle passage
"""
def __init__(self, data):
self.id = data["id"]
self.location = data["location"]
self.destination = data["destination"]
self.is_realtime = data["realtime"]
self.wait_time = data["wait_time"]
self.wait_time_text = data["wait_time_human"]
self.arrival = data["arrival"]
return Vehicle(self.ve[vehicle])
return Line(self.data)