Merge pull request #150 from FLamparski/trains

Add Trains app

trains/api.py

@@ -0,0 +1,23 @@
+import http
+import ujson
+from tilda import LED
+
+API_URL = "https://huxley.apphb.com/all/{}?expand=true&accessToken=D102521A-06C6-44C9-8693-7A0394C757EF"
+
+def get_trains(station_code='LBG'):
+    print('trains/api: Getting trains for {}'.format(station_code))
+    station_data = None
+
+    LED(LED.RED).on()  # Red for total get_trains
+    try:
+        station_json = http.get(API_URL.format(
+            station_code)).raise_for_status().content
+        LED(LED.GREEN).on()  # Green for parsing
+        station_data = ujson.loads(station_json)
+    except Exception as e:
+        print('Error:')
+        print(e)
+
+    LED(LED.RED).off()
+    LED(LED.GREEN).off()
+    return station_data

trains/departure_screen.py

@@ -0,0 +1,117 @@
+import sleep
+import ugfx
+import database
+from time import time
+from homescreen import time_as_string
+from tilda import Buttons
+from trains.screen import Screen, S_CONTINUE, S_TO_SETTINGS, S_EXIT
+from trains.api import get_trains
+from trains.utils import get_departure, get_title, is_red
+
+UPDATE_INTERVAL_SECS = 30
+
+class DepartureScreen(Screen):
+    def __init__(self):
+        self.station_data = None
+        self.has_error = False
+        self.last_update = 0
+        self.should_redraw = True
+
+        self._names = None
+        self._old_names = None
+
+    def enter(self):
+        self.next_state = S_CONTINUE
+        self.station_code = database.get('trains.station_code', 'LBG')
+        self.last_update = 0
+        Buttons.enable_interrupt(
+            Buttons.BTN_A,
+            lambda t: self.set_next_state(S_TO_SETTINGS),
+            on_press=True,
+            on_release=False
+        )
+        Buttons.enable_interrupt(
+            Buttons.BTN_Menu,
+            lambda t: self.set_next_state(S_EXIT),
+            on_press=True,
+            on_release=False
+        )
+    
+    def set_next_state(self, s):
+        self.next_state = s
+
+    def update(self):
+        now = time()
+        if self.last_update < (now - UPDATE_INTERVAL_SECS):
+            print('trains/departure_screen: Updating data')
+            new_station_data = get_trains(self.station_code)
+            if new_station_data == None:
+                self.has_error = True
+                self.should_redraw = True
+            else:
+                self.station_data = new_station_data
+                self.has_error = False
+                self.should_redraw = True
+            self.last_update = now
+    
+    def tick(self):
+        self.update()
+
+        if self.should_redraw:
+            if self.station_data == None:
+                self.show_error()
+            else:
+                self.show_trains()
+        else:
+            self._destroy_old_names()
+        
+        sleep.sleep_ms(500)
+        
+        return self.next_state
+    
+    def _get_names_container(self):
+        if self._names != None:
+            self._names.hide()
+            self._old_names = self._names
+        names = ugfx.Container(0, 25, 190, 295)
+        self._names = names
+        return names
+    
+    def _destroy_old_names(self):
+        if self._old_names != None:
+            self._old_names.destroy()
+            self._old_names = None
+    def _destroy_names(self):
+        if self._names != None:
+            self._names.destroy()
+            self._names = None
+    
+    def show_trains(self):
+        ugfx.clear()
+        ugfx.area(0, 0, 240, 25, ugfx.RED if self.has_error else ugfx.GRAY)
+        title = get_title(self.station_data['locationName'], self.has_error)
+        ugfx.text(5, 5, title, ugfx.WHITE if self.has_error else ugfx.BLACK)
+        ugfx.text(195, 5, time_as_string(), ugfx.BLUE)
+        
+        names = self._get_names_container()
+        names.show()
+        row_num = 0
+        for service in self.station_data['trainServices']:
+            departure = get_departure(service)
+            if departure:
+                names.text(5, 15 * row_num, service['destination'][0]['locationName'], ugfx.BLACK)
+                ugfx.text(195, 25 + (15 * row_num), departure,ugfx.RED if is_red(service) else ugfx.BLUE)
+                row_num += 1
+
+        ugfx.display_image(0, 300, 'trains/bottom.gif')
+        self.should_redraw = False
+
+    def show_error(self):
+        ugfx.clear()
+        ugfx.text(5, 5, 'Error :(', ugfx.RED)
+        self.should_redraw = False
+
+    def exit(self):
+        self._destroy_old_names()
+        self._destroy_names()
+        Buttons.disable_all_interrupt()

trains/main.py

@@ -0,0 +1,92 @@
+"""Mini train departure board for your badge
+
+Configurable with which station you want to monitor
+"""
+___title___ = "trains"
+___license___ = "MIT"
+___dependencies___ = ["app", "sleep", "wifi", "http", "ugfx_helper"]
+___categories___ = ["Homescreens", "Other"]
+___bootstrapped___ = False
+
+
+import database
+import wifi
+import ugfx
+import app
+import sleep
+import ntp
+from tilda import Buttons, LED
+from trains import api
+from trains import screen
+from trains.departure_screen import DepartureScreen
+from trains.settings_screen import SettingsScreen
+
+def init_screen(orientation):
+    # initialize screen
+    ugfx.clear()
+    ugfx.orientation(orientation)
+    ugfx.backlight(50)
+    # show initial screen
+    # photo credit: https://www.flickr.com/photos/remedy451/8061918891
+    ugfx.display_image(0, 0, 'trains/splash.gif', 90)
+
+
+def init():
+    print('trains/main: Init')
+    ugfx.init()
+    ntp.set_NTP_time()
+    # ensure wifi connection
+    if not wifi.is_connected():
+        wifi.connect(show_wait_message=True)
+
+
+def exit():
+    print('trains/main: Exit')
+    ugfx.clear()
+    app.restart_to_default()
+
+
+app_screens = {
+    screen.SETTINGS: SettingsScreen(),
+    screen.DEPARTURES: DepartureScreen()
+}
+
+
+def get_initial_screen():
+    station_code = database.get('trains.station_code', None)
+    if station_code == None:
+        return app_screens[screen.SETTINGS]
+    return app_screens[screen.DEPARTURES]
+
+
+def run_screen(instance):
+    print('trains/main: Starting screen {}'.format(instance))
+    instance.enter()
+
+    is_running = True
+    next_screen_name = None
+    while is_running:
+        status, value = instance.tick()
+
+        if status == screen.SWITCH_SCREEN:
+            is_running = False
+            next_screen_name = value
+        elif status == screen.EXIT_APP:
+            is_running = False
+    
+    print('trains/main: Stopping screen {} (next = {})'.format(instance, next_screen_name))
+    instance.exit()
+    return next_screen_name
+
+init()
+current_screen = get_initial_screen()
+is_app_running = True
+while is_app_running:
+    init_screen(current_screen.orientation())
+    next_screen_name = run_screen(current_screen)
+
+    if next_screen_name != None:
+        current_screen = app_screens[next_screen_name]
+    else:
+        is_app_running = False
+        exit()

trains/screen.py

@@ -0,0 +1,28 @@
+CONTINUE = 1
+SWITCH_SCREEN = 2
+EXIT_APP = 3
+
+DEPARTURES = 10
+SETTINGS = 11
+
+S_CONTINUE = (CONTINUE, None)
+S_TO_SETTINGS = (SWITCH_SCREEN, SETTINGS)
+S_TO_TRAINS = (SWITCH_SCREEN, DEPARTURES)
+S_EXIT = (EXIT_APP, None)
+
+
+class Screen():
+    def __init__(self):
+        pass
+
+    def orientation(self):
+        return 90
+
+    def enter(self):
+        pass
+    
+    def tick(self):
+        return S_CONTINUE
+        
+    def exit(self):
+        pass

trains/settings_screen.py

@@ -0,0 +1,19 @@
+import database
+import ugfx
+from dialogs import prompt_text
+from trains.screen import Screen, S_CONTINUE, S_TO_TRAINS
+
+class SettingsScreen(Screen):
+    def __init__(self):
+        self.next_state = S_TO_TRAINS
+    
+    def orientation(self):
+        return 270
+    
+    def tick(self):
+        with database.Database() as db:
+            crs = prompt_text('Enter your station\'s CRS code', db.get('trains.station_code', ''))
+            db.set('trains.station_code', crs)
+
+        return self.next_state
+    

trains/utils.py

@@ -0,0 +1,19 @@
+def is_red(service):
+    return service['isCancelled'] or service['etd'] != 'On time'
+
+
+def get_departure(service):
+    if service['isCancelled']:
+        return 'CANX'
+
+    if service['etd'] == 'On time':
+        return service['std']
+
+    return service['etd']
+
+
+def get_title(name, has_error):
+    if has_error:
+        return 'ERR ' + name
+
+    return name