-
Notifications
You must be signed in to change notification settings - Fork 1
/
pyrailtimes.py
197 lines (154 loc) · 6.44 KB
/
pyrailtimes.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
#!/usr/bin/env python
import curses
import datetime
import schedule
from display.board import Board
from config.config import Config
from data.data import StationData, ServiceData
import time
class PyRailTimes:
__services = dict()
__station_data = dict() # (station_name code / StationData)
__boards = dict()
__config = Config()
# Dimensions of the current terminal window
__rows = 0
__cols = 0
__board_count = 0
# Time of the last API update
__last_update = datetime.datetime.now()
def __init__(self):
for config_station in self.__config.stations:
if config_station.code not in self.__station_data:
self.__station_data[config_station.code] = StationData(config_station.code)
if config_station.code not in self.__boards:
self.__boards[config_station.code] = dict()
# TODO calculate board positions based on available terminal size
# We need to cast platform as a string i.e. in case we have platform 1a
self.__boards[config_station.code][str(config_station.platform)] = Board((self.__board_count * 10) + 1, 0)
self.__board_count += 1
self.__stdscr = curses.initscr()
self.__update_size()
def __print_header(self):
"""
Print the app title and time in the top right corner
"""
title = "pyRailTimes"
self.__stdscr.addstr(0, max(0, self.__cols - len(title)), title)
update = "Last update: {}".format(
datetime.datetime.strftime(self.__last_update, "%H:%M:%S")
)
self.__stdscr.addstr(1, max(0, self.__cols - len(update)), update)
#
# Resize handlers
#
def __update_size(self):
"""
Update the known height and width of the window
"""
self.__rows, self.__cols = self.__stdscr.getmaxyx()
def __handle_resize(self):
self.__update_size()
self.__stdscr.clear()
curses.resizeterm(self.__rows, self.__cols)
self.__stdscr.addstr(0,0,"Rows: {} Cols: {}".format(self.__rows, self.__cols))
self.__redraw()
def __redraw(self):
self.__print_header()
self.__stdscr.refresh()
for station_boards in self.__boards.values():
for board in station_boards.values():
board.redraw()
#
# Departure Board handlers
#
def __setup_schedules(self):
"""
Recurring events for the departure board
"""
schedule.every(0.5).seconds.do(self.__update_board_times)
schedule.every(0.1).seconds.do(self.__advance_board_tickers)
schedule.every(5).seconds.do(self.__advance_board_destinations)
schedule.every().minute.do(self.__update_station_data)
def __update_board_times(self):
for station_boards in self.__boards.values():
for board in station_boards.values():
board.update_time()
def __advance_board_tickers(self):
for station_boards in self.__boards.values():
for board in station_boards.values():
board.advance_ticker()
def __advance_board_destinations(self):
for station_boards in self.__boards.values():
for board in station_boards.values():
board.advance_destinations()
#
# Data update handlers
#
def __update_station_data(self):
for station_data in self.__station_data.values():
station_data.check_updates(lambda station_code: self.__on_station_update(station_code))
def __on_station_update(self, station_code):
# Update "last updated" message
self.__last_update = datetime.datetime.now()
self.__print_header()
data: StationData = self.__station_data[station_code]
boards = self.__boards[station_code]
for platform, board in boards.items():
board.set_station(data.data.location.name, platform)
self.__stdscr.refresh()
if not data.services:
board.show_no_services()
else:
# Filter by platform
platform_services = [service for service in data.services if service.location_detail.platform == platform]
if not platform_services:
board.show_no_services()
else:
board.update_services(platform_services)
first_service = platform_services[0]
if first_service.service_uid not in self.__services:
self.__update_service_data(first_service, station_code, platform)
self.__cleanup_services()
def __cleanup_services(self):
"""
Cleans up the list of service calling points so we don't have any memory issues
"""
current_services = dict()
for service_uid in self.__services.keys():
for station_data in self.__station_data.values():
if station_data.services:
if station_data.services[0].service_uid == service_uid:
current_services[service_uid] = self.__services[service_uid]
self.__services = current_services
def __update_service_data(self, service, station_code, platform):
service_data = ServiceData(
service.service_uid,
service.run_date,
lambda data: self.__on_service_update(data, station_code, platform)
)
self.__services[service.service_uid] = service_data
def __on_service_update(self, service_data, station_code, platform):
board = self.__boards[station_code][platform]
board.update_service_calling_points(service_data.calling_points, station_code)
#
# Main program loop
#
def mainloop(self):
for station_boards in self.__boards.values():
for platform_board in station_boards.values():
platform_board.draw_box()
platform_board.update_time()
for station_data in self.__station_data.values():
station_data.check_updates(lambda station_code: self.__on_station_update(station_code))
self.__setup_schedules()
while True:
# Check if screen was re-sized (True or False)
resize = curses.is_term_resized(self.__rows, self.__cols)
if resize is True:
self.__handle_resize()
schedule.run_pending()
time.sleep(0.1)
# Entry point
if __name__ == '__main__':
curses.wrapper(PyRailTimes().mainloop())