|
1 | | -""" |
2 | | -NextBus class -- handles NextBus server queries and infers |
3 | | -arrival times based on last-queried predictions and elapsed time. |
4 | | -""" |
5 | | - |
6 | | -# pylint: disable=bare-except, too-many-instance-attributes, too-many-arguments |
7 | | - |
8 | | -import re |
9 | | -import time |
10 | | - |
11 | | -class NextBus(): |
12 | | - """ Class to handle NextBus prediction times for one route & stop. |
13 | | - """ |
14 | | - |
15 | | - def __init__(self, network, agency, route, stop, data=None, |
16 | | - max_predictions=3, minimum_time=300): |
17 | | - """ Constructor expects a Requests-capable Network object, |
18 | | - strings for transit agency, route, and stop ID, plus optional |
19 | | - per-stop data as defined by an application (e.g. text |
20 | | - description, but could be an object or tuple of data, or None, |
21 | | - or whatever's needed by the app), limits for the maximum number |
22 | | - of future arrivals to predict (limited to server response, |
23 | | - typically 5) and minimum time below which arrivals are not shown |
24 | | - (to discourage unsafe bus-chasing). |
25 | | - """ |
26 | | - self.network = network |
27 | | - self.agency = agency |
28 | | - self.route = route |
29 | | - self.stop = stop |
30 | | - self.data = data |
31 | | - self.max_predictions = max_predictions |
32 | | - self.minimum_time = minimum_time |
33 | | - self.predictions = [] |
34 | | - self.last_query_time = -1000 |
35 | | - |
36 | | - def fetch(self): |
37 | | - """ Contact NextBus server and request predictions for one |
38 | | - agency/route/stop. |
39 | | - """ |
40 | | - try: |
41 | | - url = ('http://webservices.nextbus.com/service/publicXMLFeed?' + |
42 | | - 'command=predictions&a=' + self.agency + |
43 | | - '&r=' + self.route + '&s=' + self.stop) |
44 | | - response = self.network.requests.get(url) |
45 | | - if response.status_code == 200: |
46 | | - string = response.text |
47 | | - self.last_query_time = time.monotonic() |
48 | | - self.predictions = [] |
49 | | - while len(self.predictions) < self.max_predictions: |
50 | | - # CircuitPython version of re library doesn't have findall. |
51 | | - # Search for first instance of seconds="N" string and then |
52 | | - # increment the string position based on match length. |
53 | | - match = re.search('seconds=\"[0-9]*', string) |
54 | | - if match: |
55 | | - seconds = int(match.group(0)[9:]) # Remove 'seconds="' |
56 | | - if seconds >= self.minimum_time: |
57 | | - self.predictions.append(seconds) |
58 | | - string = string[match.end():] |
59 | | - else: |
60 | | - break |
61 | | - self.predictions.sort() |
62 | | - except: |
63 | | - # If server query fails, we can keep extrapolating from the |
64 | | - # last set of predictions and try query again on next pass. |
65 | | - pass |
66 | | - |
67 | | - def predict(self): |
68 | | - """ Extrapolate predictions based on last values queried from |
69 | | - NextBus server and time elapsed since last query. Predictions |
70 | | - are returned as a list of integer seconds values. |
71 | | - """ |
72 | | - times = [] |
73 | | - for predict in self.predictions: |
74 | | - seconds = predict - (time.monotonic() - self.last_query_time) |
75 | | - if seconds >= self.minimum_time: |
76 | | - times.append(seconds) |
77 | | - return times |
| 1 | +""" |
| 2 | +NextBus class -- handles NextBus server queries and infers |
| 3 | +arrival times based on last-queried predictions and elapsed time. |
| 4 | +""" |
| 5 | + |
| 6 | +# pylint: disable=bare-except, too-many-instance-attributes, too-many-arguments |
| 7 | + |
| 8 | +import re |
| 9 | +import time |
| 10 | + |
| 11 | +class NextBus(): |
| 12 | + """ Class to handle NextBus prediction times for one route & stop. |
| 13 | + """ |
| 14 | + |
| 15 | + def __init__(self, network, agency, route, stop, data=None, |
| 16 | + max_predictions=3, minimum_time=300): |
| 17 | + """ Constructor expects a Requests-capable Network object, |
| 18 | + strings for transit agency, route, and stop ID, plus optional |
| 19 | + per-stop data as defined by an application (e.g. text |
| 20 | + description, but could be an object or tuple of data, or None, |
| 21 | + or whatever's needed by the app), limits for the maximum number |
| 22 | + of future arrivals to predict (limited to server response, |
| 23 | + typically 5) and minimum time below which arrivals are not shown |
| 24 | + (to discourage unsafe bus-chasing). |
| 25 | + """ |
| 26 | + self.network = network |
| 27 | + self.agency = agency |
| 28 | + self.route = route |
| 29 | + self.stop = stop |
| 30 | + self.data = data |
| 31 | + self.max_predictions = max_predictions |
| 32 | + self.minimum_time = minimum_time |
| 33 | + self.predictions = [] |
| 34 | + self.last_query_time = -1000 |
| 35 | + |
| 36 | + def fetch(self): |
| 37 | + """ Contact NextBus server and request predictions for one |
| 38 | + agency/route/stop. |
| 39 | + """ |
| 40 | + try: |
| 41 | + url = ('http://webservices.nextbus.com/service/publicXMLFeed?' + |
| 42 | + 'command=predictions&a=' + self.agency + |
| 43 | + '&r=' + self.route + '&s=' + self.stop) |
| 44 | + response = self.network.requests.get(url) |
| 45 | + if response.status_code == 200: |
| 46 | + string = response.text |
| 47 | + self.last_query_time = time.monotonic() |
| 48 | + self.predictions = [] |
| 49 | + while len(self.predictions) < self.max_predictions: |
| 50 | + # CircuitPython version of re library doesn't have findall. |
| 51 | + # Search for first instance of seconds="N" string and then |
| 52 | + # increment the string position based on match length. |
| 53 | + match = re.search('seconds=\"[0-9]*', string) |
| 54 | + if match: |
| 55 | + seconds = int(match.group(0)[9:]) # Remove 'seconds="' |
| 56 | + if seconds >= self.minimum_time: |
| 57 | + self.predictions.append(seconds) |
| 58 | + string = string[match.end():] |
| 59 | + else: |
| 60 | + break |
| 61 | + self.predictions.sort() |
| 62 | + except: |
| 63 | + # If server query fails, we can keep extrapolating from the |
| 64 | + # last set of predictions and try query again on next pass. |
| 65 | + pass |
| 66 | + |
| 67 | + def predict(self): |
| 68 | + """ Extrapolate predictions based on last values queried from |
| 69 | + NextBus server and time elapsed since last query. Predictions |
| 70 | + are returned as a list of integer seconds values. |
| 71 | + """ |
| 72 | + times = [] |
| 73 | + for predict in self.predictions: |
| 74 | + seconds = predict - (time.monotonic() - self.last_query_time) |
| 75 | + if seconds >= self.minimum_time: |
| 76 | + times.append(seconds) |
| 77 | + return times |
0 commit comments