Last active
February 7, 2019 22:43
-
-
Save jhsu/8282ffc09aa3380ee09fe2c418d5275d to your computer and use it in GitHub Desktop.
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
| import os | |
| import requests | |
| import time | |
| import datetime | |
| import sys | |
| import signal | |
| from papirus import PapirusComposite | |
| screen = PapirusComposite(False) | |
| screen.RemoveText("Status") | |
| def parseTime(num): | |
| return datetime.datetime.fromtimestamp(round(num / 1000)) | |
| def minutesDiff(sched, est): | |
| if est: | |
| return ((parseTime(est) - datetime.datetime.now()).total_seconds()) / 60 | |
| else: | |
| return ((parseTime(sched) - datetime.datetime.now()).total_seconds()) / 60 | |
| def get_next_stops(stop=3940): | |
| url = "https://developer.trimet.org/ws/v2/arrivals/locIDs/%s/appID/%s" % (stop, os.environ['APP_ID']) | |
| while True: | |
| resp = requests.get(url) | |
| yield resp.json().get('resultSet').get('arrival') | |
| def displayArrivals(display, displayed=dict(), arrivals=[]): | |
| height = 20 | |
| for i,arrival in enumerate(arrivals): | |
| minutes, name = arrival['minsAway'], arrival['name'] | |
| if displayed.get(str(i)): | |
| display.UpdateText(str(i), '%s | %s' % (minutes, name)) | |
| else: | |
| display.AddText('%s | %s' % (minutes, name), 0, height * i, Id=str(i)) | |
| displayed[str(i)] = True | |
| display.WriteAll() | |
| def update_display_func(display): | |
| store = dict() | |
| displayed = dict() | |
| def update_display(arrivals): | |
| changed = False; | |
| for arrival in arrivals: | |
| id, name, scheduled, estimated = arrival.get('id'), arrival.get('shortSign'), arrival.get('scheduled'), arrival.get('estimated') | |
| minsAway = round(minutesDiff(scheduled, estimated)) | |
| arrivalInfo = { | |
| 'id': id, | |
| 'name': name, | |
| 'estimated': estimated, | |
| 'scheduled': scheduled, | |
| 'minsAway': minsAway, | |
| } | |
| if store.get(arrivalInfo['id']): | |
| storedArrival = store.get(arrivalInfo['id']) | |
| if minsAway != storedArrival['minsAway']: | |
| # trigger update if time estimate | |
| changed = True | |
| else: | |
| # trigger update if new arrival | |
| changed = True | |
| store[arrivalInfo['id']] = arrivalInfo | |
| arrivalList = list(map(lambda i: store.get(i.get('id')), arrivals)) | |
| if (changed): | |
| displayArrivals(display, displayed, arrivalList) | |
| return update_display | |
| def set_offline_func(display): | |
| def set_offline(*args): | |
| display.AddText("X", 5 * 20, 5 * 20, Id="Status") | |
| display.WriteAll() | |
| sys.exit() | |
| return set_offline | |
| exiter = set_offline_func(screen) | |
| signal.signal(signal.SIGTERM, exiter) | |
| # update_display = update_display_func('') | |
| update_display = update_display_func(screen) | |
| for arrivals in get_next_stops(3940): | |
| update_display(arrivals) | |
| time.sleep(30) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment