Created
October 8, 2022 16:07
-
-
Save netsensei/ae882bf795e6d6b8cbda8fd3b8b03354 to your computer and use it in GitHub Desktop.
STIB/MIVB waiting times for a set of stops as a CLI script
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/python3 | |
# | |
# run it: | |
# STIBKEY=<apikey> ./stib.py | |
import os | |
import requests | |
import json | |
import datetime | |
from datetime import timedelta | |
from datetime import datetime | |
import time | |
import pytz | |
class PassingTime: | |
def __init__(self, passing): | |
if 'destination' in passing: | |
self.destination = passing['destination']['nl'] | |
else: | |
self.destination = "UNKOWN" | |
self.expected_arrival_time = self.waiting_time(self.expected_arrival_time) | |
def waiting_time(self, arrival_time): | |
isodate = datetime.fromisoformat(arrival_time) | |
now = datetime.now(pytz.timezone("Europe/Brussels")) | |
diff = isodate - now | |
seconds = diff.total_seconds() | |
m, s = divmod(seconds, 60) | |
m = int(m) | |
s = int(s) | |
return f'{m:02d}m {s:02d}s' | |
class Record: | |
def __init__(self, record): | |
self.line = record['lineid'] | |
self.stop = record['pointid'] | |
self.passing_times = self.parse_passing_times(record['passingtimes']) | |
def parse_passing_times(self, passing_times): | |
result = [] | |
passing_times = json.loads(passing_times) | |
for pt in passing_times: | |
result.append(PassingTime(pt)) | |
return result | |
class TableView: | |
def __init__(self, records): | |
self.register = {} | |
for record in records: | |
for pt in record.passing_times: | |
key = "s%s_l%s_%s"%(record.stop, record.line, pt.destination) | |
if key in self.register: | |
self.register[key].append({ | |
'destination': pt.destination, | |
'arrival': pt.expected_arrival_time, | |
}) | |
else: | |
self.register[key] = [{ | |
'destination': pt.destination, | |
'arrival': pt.expected_arrival_time, | |
}] | |
def display(self): | |
for direction, vals in self.register.items(): | |
shown = False | |
for val in vals: | |
if shown == False: | |
print("%s %s" % (direction.ljust(30, ' '), val['arrival'])) | |
shown = True | |
else: | |
print("%s %s" % (''.ljust(30, ' '), val['arrival'])) | |
def Main(): | |
apikey = os.environ.get("STIBKEY") | |
if apikey == None: | |
print("No STIB key set. use 'export STIBKEY='") | |
os.exit() | |
http_address="https://data.stib-mivb.be/api/v2/catalog/datasets/waiting-time-rt-production/records?where=pointid%3D5151%20or%20pointid%3D5120%20or%20pointid%3D2734%20or%20pointid%3D2735&limit=20&offset=0&timezone=UTC&apikey=" + apikey | |
response = requests.get(url=http_address) | |
data = json.loads(response.text) | |
records = [] | |
for raw in data['records']: | |
records.append(Record(raw['record']['fields'])) | |
view = TableView(records) | |
view.display() | |
# Gotime! | |
Main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment