clarify things

This commit is contained in:
mdeboute 2022-05-08 13:17:07 +02:00
parent aed901c0c0
commit 5437def293
16 changed files with 35 additions and 122 deletions

7
.gitignore vendored
View File

@ -1,10 +1,3 @@
# Byte-compiled Python # Byte-compiled Python
__pycache__/ __pycache__/
*.pyc *.pyc
# Poetry venv directories and files
.venv/
poetry.lock
# Keys and tokens of the Twitter API
credentials.py

View File

@ -1,3 +1,13 @@
# info_tbm # What's this?
A twitter bot to know if my tram will arrive soon ⏰ This is a set of libraries to gather real time data from TBM.
They all work around [the dynamic map](https://www.infotbm.com/fr/plans/plan-dynamique/).
## What's TBM?
TBM is a French public transport company in Bordeaux.
## Licence
These libraries are under GNU GPL v3. See [license](LICENSE) for more details.

Binary file not shown.

Before

Width:  |  Height:  |  Size: 76 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 22 KiB

View File

@ -1,16 +0,0 @@
[tool.poetry]
name = "info_tbm"
version = "0.1.0"
description = ""
authors = ["mdeboute <martin.deboute@gmail.com>"]
[tool.poetry.dependencies]
python = "^3.9"
gmplot = "^1.4.1"
tweepy = "^4.8.0"
[tool.poetry.dev-dependencies]
[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"

View File

@ -1,37 +0,0 @@
import tweepy
from config import *
import time
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger()
def check_mentions(api, keywords, since_id):
logger.info("Retrieving mentions")
new_since_id = since_id
for tweet in tweepy.Cursor(api.mentions_timeline, since_id=since_id).items():
new_since_id = max(tweet.id, new_since_id)
if tweet.in_reply_to_status_id is not None:
continue
if any(keyword in tweet.text.lower() for keyword in keywords):
logger.info(f"Answering to {tweet.user.name}")
api.update_status(
status="Please reach us via DM",
in_reply_to_status_id=tweet.id,
auto_populate_reply_metadata=True,
)
return new_since_id
def main():
api = create_api()
since_id = 1
while True:
since_id = check_mentions(api, ["help", "support"], since_id)
logger.info("Waiting...")
time.sleep(10)
if __name__ == "__main__":
main()

View File

@ -1,19 +0,0 @@
import tweepy
import logging
from credentials import *
logger = logging.getLogger()
def create_api():
auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
api = tweepy.API(auth, wait_on_rate_limit=True)
try:
api.verify_credentials()
except Exception as e:
logger.error("Error creating API", exc_info=True)
raise e
logger.info("API created")
return api

View File

@ -1,7 +1,7 @@
import sys import sys
from tbm_api.stop_area import * from stop_area import *
from tbm_api.stop import * from stop import *
from tbm_api.stop_route import StopRoute from stop_route import StopRoute
from datetime import datetime from datetime import datetime

View File

@ -1,8 +1,8 @@
from tbm_api.utils import get_data_from_json from utils import get_data_from_json
from urllib.parse import quote from urllib.parse import quote
from tbm_api.stop_point import StopPoint from stop_point import StopPoint
from re import search from re import search
from tbm_api.route import Route from route import Route
INFO_URL = "https://ws.infotbm.com/ws/1.0/network/stoparea-informations/%s" INFO_URL = "https://ws.infotbm.com/ws/1.0/network/stoparea-informations/%s"

View File

@ -1,4 +1,4 @@
from tbm_api.utils import get_data_from_json from utils import get_data_from_json
from urllib.parse import quote from urllib.parse import quote

View File

@ -1,6 +1,6 @@
from tbm_api.utils import get_data_from_json, hms2seconds from utils import get_data_from_json, hms2seconds
from time import time from time import time
from tbm_api.vehicle import Vehicle from vehicle import Vehicle
SCHEDULE_URL = "https://ws.infotbm.com/ws/1.0/get-realtime-pass/%d/%s" SCHEDULE_URL = "https://ws.infotbm.com/ws/1.0/get-realtime-pass/%d/%s"
@ -8,7 +8,7 @@ SCHEDULE_URL = "https://ws.infotbm.com/ws/1.0/get-realtime-pass/%d/%s"
class Line: class Line:
""" """
Information on the line served at a stop Information on the line served at a stop.
""" """
def __init__(self, vehicles): def __init__(self, vehicles):
@ -41,7 +41,7 @@ class StopRoute:
def update(self, auto=False): def update(self, auto=False):
""" """
Update data Update data.
""" """
data = get_data_from_json(SCHEDULE_URL % (self.number, self.line)) data = get_data_from_json(SCHEDULE_URL % (self.number, self.line))
@ -80,7 +80,7 @@ class StopRoute:
def data_age(self): def data_age(self):
""" """
Returns the age of the data Returns the age of the data.
""" """
return time() - self.last_update return time() - self.last_update

View File

@ -5,7 +5,7 @@ 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()
@ -17,7 +17,7 @@ def get_data_from_json(url):
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:

View File

@ -1,33 +1,17 @@
""" """
Provides all info about stations Provides all info about stations.
""" """
from tbm_api.utils import get_data_from_json from utils import get_data_from_json
from time import time from time import time
vcub_url = "https://ws.infotbm.com/ws/1.0/vcubs" vcub_url = "https://ws.infotbm.com/ws/1.0/vcubs"
class Vcub: class Vcub:
""" """
Retrieves information from stations Retrieves information from stations.
Data format, as returned by the infotbm website:
{
lists: [
{
'id': station number,
'name': str,
'connexionState': 'CONNECTED' if in service 'DISCONNECTED' otherwise,
'typeVlsPlus': 'VLS_PLUS' if + 'PAS_VLS_PLUS' otherwise,
'nbPlaceAvailable': 33,
'nbBikeAvailable': 0
'nbElectricBikeAvailable': 6
'latitude': float,
'longitude': float,
},
]
}
""" """
def __init__( def __init__(
@ -49,9 +33,7 @@ class Vcub:
def update(self, auto=False, data=None): def update(self, auto=False, data=None):
""" """
Updates data 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: if data is None or type(data) != dict:
@ -77,14 +59,14 @@ class Vcub:
def data_age(self): def data_age(self):
""" """
Computes the data's age Computes the data's age.
""" """
return time() - self.last_update return time() - self.last_update
def get_names(self): def get_names(self):
""" """
Returns all names in a dict with id as data Returns all names in a dict with id as data.
""" """
r = {} r = {}
@ -94,7 +76,7 @@ class Vcub:
def get_locations(self): def get_locations(self):
""" """
Returns all locations in a dict with id as data Returns all locations in a dict with id as data.
""" """
r = {} r = {}
@ -104,7 +86,7 @@ class Vcub:
def get_by_id(self, id): def get_by_id(self, id):
""" """
Returns a station by its id Returns a station by its id.
""" """
class Station: class Station: