Created
November 27, 2014 11:24
-
-
Save icedraco/2b736541d10748bac975 to your computer and use it in GitHub Desktop.
An SL4A (Android) Python script that polls the MetroDan station page and keeps track of ETAs for specific routes. The script notifies of ETA changes using Android's speech synthesis API.
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
###--# MetroDan Bus Tracker / Announcer 0.5.1a | |
# (c) 2013-2014 by IceDragon - All rights reserved | |
# | |
# This script Polls MetroDan "real-time" bus website and keeps the user | |
# informed of the buses they are interested in via Speech To Text. | |
# | |
# This script is still work-in-progress. | |
# | |
# Author: IceDragon <icedragon at quickfox org> | |
# | |
# Changelog: | |
# 0.5.1a - Indentation bugfix | |
# 0.5a - Added conectivity issues handler that avoids crashes | |
# 0.4a - Added "special mode" for old city station and route 8 | |
import android | |
import urllib2 | |
import re | |
import time | |
import sys | |
###--# Pseudo-Constants #--#################################################### | |
RE_TIME_TABLE = re.compile("<tr><td>([0-9]+)</td><td>([^<]+)</td></tr>") | |
LOOKUP_URL_FORMAT = 'http://m.ontime3g.com/?stop=%d&company=md' | |
HEBREW_WORDS = { | |
" \xd7\x93\xd7\xa7\xd7\x95\xd7\xaa": " mins", | |
" \xd7\x93\xd7\xa7\xd7\x94": " min", | |
" \xd7\x95": " and", | |
" \xd7\xa9\xd7\xa2\xd7\x94": " hour" | |
} | |
TMP_STOP_IDS = { | |
'Hapoalim': 11121, | |
'Old City': 11126, | |
'BGU Rear': 11214, | |
'BGU AUX': 15332, | |
'BGU Stop': 13949, | |
'BGU Soroka': 13518 | |
} | |
BUS_ID_LIST = { | |
# Default | |
None: [24, 34, 7, 8], | |
# Old city - ignore other routes | |
11126: [8] | |
} | |
###--# Classes #--############################################################# | |
class Station: | |
def __init__(self, station_id, name=None, filter=None): | |
if not name: | |
name = str(station_id) | |
self.id = int(station_id) | |
self.name = str(name) | |
self.buses = {} | |
self.last_scan = 0 | |
self.next_id = 0 | |
self.filter = filter | |
def __str__(self): | |
return self.name | |
def __repr__(self): | |
return "<Station #%d>" % self.id | |
def getNextRoute(self): | |
return int(self.next_id) | |
def getEta(self, busNo): | |
return self.buses[busNo] | |
def getRouteList(self): | |
return self.buses.keys() | |
def getDataAge(self): | |
return time.time() - self.last_scan | |
def update(self): | |
"""Updates information about buses from the web""" | |
url = LOOKUP_URL_FORMAT % self.id | |
# Get the data as HTML from the website | |
data = urllib2.urlopen(url).read() | |
# Extract route information from the HTML data | |
raw_routes = RE_TIME_TABLE.findall(data) | |
routes = map(handleHebrewWords, raw_routes) | |
# Process and store acquired information | |
min = 9999 | |
for route in routes: | |
(lineID, time_str) = route | |
lineID = int(lineID) | |
# Ignore uninteresting routes | |
if self.filter and not self.filter.accept(lineID): | |
continue | |
eta_mins = translateTimeString(time_str) | |
self.buses[lineID] = eta_mins | |
if eta_mins > 0 and eta_mins < min: | |
self.next_id = lineID | |
min = eta_mins | |
self.last_scan = time.time() | |
return self | |
class BusFilter: | |
def __init__(self, route_list = []): | |
self.__routes = [] | |
for route_id in route_list: | |
self.add(int(route_id)) | |
def __str__(self): | |
return ", ".join(self.__routes) | |
def accept(self, route_id): | |
return route_id in self.__routes | |
def add(self, id): | |
self.__routes += [id] | |
###--# Standalone Helper Functions #--######################################### | |
def translateTimeString(timeString): | |
last_num = -1 | |
hour_mult = 1 | |
hours = 1 | |
mins = 0 | |
for token in timeString.split(' '): | |
if token.isdigit(): | |
last_num = int(token) | |
elif token == "min": | |
mins = 1 | |
elif token == "hour": | |
if last_num > 0: | |
hours = last_num | |
last_num = -1 | |
hour_mult = 60 | |
elif token == "mins": | |
mins = last_num | |
last_num = -1 | |
return hours * hour_mult + mins | |
def handleHebrewWords(route): | |
global HEBREW_WORDS | |
(line, eta) = route | |
for w in HEBREW_WORDS.keys(): | |
eta = eta.replace(w, HEBREW_WORDS[w]) | |
return (line, eta) | |
###--# Standalone Helper Functions #--######################################### | |
def say(msg): | |
global DROID | |
print msg | |
DROID.ttsSpeak(msg) | |
def promptStationID(): | |
global TMP_STOP_IDS, DROID | |
option_list = TMP_STOP_IDS.keys() | |
DROID.dialogCreateAlert("Select Station") | |
DROID.dialogSetSingleChoiceItems(option_list) | |
DROID.dialogSetPositiveButtonText("Select") | |
DROID.dialogShow() | |
response = DROID.dialogGetResponse() | |
key_index = DROID.dialogGetSelectedItems().result[0] | |
return TMP_STOP_IDS[ option_list[key_index] ] | |
def main(argv): | |
global TMP_STOP_IDS, DROID, BUS_ID_LIST | |
DROID = android.Android() | |
station_id = promptStationID() | |
DROID.wakeLockAcquirePartial() | |
# If this station has a specific ID list, use that instead of the defaults | |
id_set = None | |
if station_id in BUS_ID_LIST: | |
id_set = station_id | |
if not id_set: | |
mode_str = "standard" | |
else: | |
mode_str = "special" | |
id_list = BUS_ID_LIST[id_set] | |
say("Tracking established in %s mode" % mode_str) | |
# Find the station and prepare for data fetching | |
s = Station(station_id, "", BusFilter(id_list)) | |
while True: | |
eta = 0 | |
try: | |
s.update() | |
except: | |
say("Station update failed"); | |
else: | |
if s.getRouteList() == []: | |
say("Station tracking lost") | |
else: | |
next_id = s.getNextRoute() | |
eta = s.getEta(next_id) | |
say("Bus %d E T A: %d minutes" % (next_id, eta)) | |
sleep_time = 30 | |
if eta > 10: | |
sleep_time = 120 | |
time.sleep(sleep_time) | |
return 0 | |
main([]) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment