This is getting too big, so commiting now: More Tilda mk4 changes

.development/firmware_update.py

@@ -20,7 +20,8 @@ def firmware_update(verbose):
             print("We couldn't find a DFU enabled badge. Please check the following:")
             print("")
             print("1) Your badge is plugged into this computer via USB")
-            print("2) Your badge is in DFU mode. You can tell by a small, red flashing light at the back")
+            print("2) The switch underneath the screen at the back of the badge is set to 'on'")
+            print("3) Your badge is in DFU mode. You can tell by a small, red flashing light at the back")
             print("")
             print("To put your badge into DFU mode (or if you're unsure whether it really is) you need to")
             print("press the joystick to the right while pressing the reset button at the back.")
@@ -28,7 +29,7 @@ def firmware_update(verbose):
             print("After that, please try this script again.")
             return
 
-        print("Downloading newest firmware: ", end="")
+        print("Downloading newest firmware: ", end="", flush=True)
         with urllib.request.urlopen(url) as response:
             with open(temp_path, 'wb') as tmp_file:
                 shutil.copyfileobj(response, tmp_file)

.development/pyboard.py

@@ -240,7 +240,7 @@ class Pyboard:
             delayed = False
             for attempt in range(wait + 1):
                 try:
-                    self.serial = serial.Serial(device, baudrate=baudrate, interCharTimeout=1)
+                    self.serial = serial.Serial(device, baudrate=baudrate, interCharTimeout=1, timeout=1, write_timeout=1)
                     break
                 except (OSError, IOError): # Py2 and Py3 have different errors
                     if wait == 0:
@@ -282,9 +282,8 @@ class Pyboard:
                 time.sleep(0.01)
         return data
 
-    def enter_raw_repl(self):
+    def enter_raw_repl(self, retry_count = 2):
         self.serial.write(b'\r\x03\x03') # ctrl-C twice: interrupt any running program
-
         # flush input (without relying on serial.flushInput())
         n = self.serial.inWaiting()
         while n > 0:
@@ -295,6 +294,8 @@ class Pyboard:
         data = self.read_until(1, b'raw REPL; CTRL-B to exit\r\n>')
         if not data.endswith(b'raw REPL; CTRL-B to exit\r\n>'):
             print(data)
+            if retry_count:
+                self.enter_raw_repl(retry_count - 1)
             raise PyboardError('could not enter raw repl')
 
         self.serial.write(b'\x04') # ctrl-D: soft reset

.development/pyboard_util.py

@@ -6,7 +6,7 @@ _pyb = None
 def get_pyb(args):
     global _pyb
     if not _pyb:
-        print("Connected to badge:", end="")
+        print("Connected to badge:", end="", flush=True)
         if not args.device:
             args.device = find_tty()
 
@@ -27,11 +27,9 @@ def close_pyb():
 
 def stop_badge(args, verbose):
     pyb = get_pyb(args)
-    if verbose:
+    print("Stopping running app:", end="", flush=True)
-        print("Stopping running app:", end="")
     write_command(pyb, b'\r\x03\x03') # ctrl-C twice: interrupt any running program
-    if verbose:
+    print(" DONE")
-        print(" DONE")
 
 def write_command(pyb, command):
     flush_input(pyb)
@@ -47,7 +45,7 @@ def flush_input(pyb):
 def soft_reset(args, verbose = True):
     pyb = get_pyb(args)
     if verbose:
-        print("Soft reboot:", end="")
+        print("Soft reboot:", end="", flush=True)
     write_command(pyb, b'\x04') # ctrl-D: soft reset
     data = pyb.read_until(1, b'soft reboot\r\n')
     if data.endswith(b'soft reboot\r\n'):
@@ -76,7 +74,7 @@ def run(args, paths, verbose=True):
     pyb = get_pyb(args)
 
     if verbose:
-        print("Preparing execution:", end="")
+        print("Preparing execution:", end=" ", flush=True)
     # run any command or file(s) - this is mostly a copy from pyboard.py
     if len(paths):
         # we must enter raw-REPL mode to execute commands

.development/sync.py

@@ -31,7 +31,7 @@ def sync(storage, patterns, resources, verbose):
         if not found:
             print("WARN: No resources to copy found for pattern %s" % patterns)
     if not verbose:
-        print("Copying %s files: " % len(paths), end="")
+        print("Copying %s files: " % len(paths), end="", flush=True)
     for path in paths:
         if not path:
             continue
@@ -41,7 +41,7 @@ def sync(storage, patterns, resources, verbose):
         if verbose:
             print("Copying %s..." % rel_path)
         else:
-            print(".", end="")
+            print(".", end="", flush=True)
 
         target = os.path.join(storage, rel_path)
         target_dir = os.path.dirname(target)

badge_store/main.py

@@ -6,21 +6,21 @@ To publish apps use https://badge.emfcamp.org"""
 
 ___license___      = "MIT"
 ___title___        = "Badge Store"
-___dependencies___ = ["app", "badge_store", "dialogs"]
+___dependencies___ = ["app", "badge_store", "dialogs", "ugfx_helper"]
 ___categories___   = ["System"]
 ___bootstrapped___ = True
 
-import ugfx
+import ugfx_helper
 import os
 import wifi
 from dialogs import *
 import app
 from lib.badge_store import BadgeStore
 
-ugfx.init()
-
 ### VIEWS ###
 
+ugfx_helper.init()
+
 store = BadgeStore()
 title = "TiLDA Badge Store"
 
@@ -89,5 +89,6 @@ def main_menu():
         else:
             app.restart_to_default()
 
+wifi.connect(show_wait_message=True)
 main_menu()
 #show_app("launcher")

lib/badge_store.py

@@ -70,12 +70,19 @@ class Installer:
         count = 0
         while get_hash(TEMP_FILE) != self.hash:
             count += 1
+            print(count)
             if count > 5:
-                os.remove(TEMP_FILE)
+                try:
+                    os.remove(TEMP_FILE)
+                except:
+                    pass
                 raise OSError("Aborting download of %s after 5 unsuccessful attempts" % self.path)
             try:
+                print("download ", self.url, self.params)
                 get(self.url, params=self.params).raise_for_status().download_to(TEMP_FILE)
-            except OSError:
+            except OSError as e:
+                if "404" in str(e):
+                    raise e
                 pass
         try:
             os.remove(self.path)

lib/buttons.py

@@ -1,41 +1,24 @@
-"""Convenience methods for dealing with the TiLDA buttons"""
+"""Convenience methods for dealing with the TiLDA buttons
+
+Pins are decined in tilda.Buttons.BTN_XYZ:
+BTN_0 - BTN_9, BTN_Hash, BTN_Star
+BTN_A, BTN_B
+BTN_Call, BTN_End
+BTN_Menu
+JOY_Center, JOY_Down, JOY_Left, JOY_Right, JOY_Up
+"""
 
 ___license___ = "MIT"
 
-import machine, time
+import machine, time, tilda
 
-CONFIG = {
+# Convenience
-    "JOY_UP": [1, machine.Pin.PULL_DOWN],
+Buttons = tilda.Buttons
-    "JOY_DOWN": [2, machine.Pin.PULL_DOWN],
-    "JOY_RIGHT": [4, machine.Pin.PULL_DOWN],
-    "JOY_LEFT": [3, machine.Pin.PULL_DOWN],
-    "JOY_CENTER": [0, machine.Pin.PULL_DOWN],
-    "BTN_MENU": [5, machine.Pin.PULL_UP]
-}
-# todo: port expander
 
-_tilda_pins = {}
-_tilda_interrupts = {}
 _tilda_bounce = {}
 
-def _get_pin(button):
-    if button not in _tilda_pins:
-        raise ValueError("Please call button.init() first before using any other button functions")
-    return _tilda_pins[button]
-
-def init(buttons = CONFIG.keys()):
-    """Inits all pins used by the TiLDA badge"""
-    global _tilda_pins
-    for button in buttons:
-        _tilda_pins[button] = machine.Pin(CONFIG[button][0], machine.Pin.IN)
-        _tilda_pins[button].init(machine.Pin.IN, CONFIG[button][1])
-
 def is_pressed(button):
-    pin = _get_pin(button)
+    return tilda.Buttons.is_pressed(button)
-    if pin.pull() == machine.Pin.PULL_DOWN:
-        return pin.value() > 0
-    else:
-        return pin.value() == 0
 
 def is_triggered(button, interval = 30):
     """Use this function if you want buttons as a trigger for something in a loop
@@ -52,22 +35,30 @@ def is_triggered(button, interval = 30):
                 return False # The button might have bounced back to high
 
         # Wait for a while to avoid bounces to low
-        machine.sleep_ms(interval)
+        time.sleep_ms(interval)
 
         # Wait until button is released again
         while is_pressed(button):
-            machine.sleep_ms(1)
+            time.sleep_ms(1)
 
         _tilda_bounce[button] = time.ticks_ms() + interval
         return True
 
+
+
+# The following functions might not work
+
+
 def has_interrupt(button):
-    global _tilda_interrupts
+    return False;
-    _get_pin(button)
+
-    if button in _tilda_interrupts:
+    # todo: re-enable
-        return True
+    #global _tilda_interrupts
-    else:
+    #_get_pin(button)
-        return False
+    #if button in _tilda_interrupts:
+    #    return True
+    #else:
+    #    return False
 
 
 def enable_interrupt(button, interrupt, on_press = True, on_release = False):

lib/dialogs.py

@@ -32,7 +32,7 @@ def prompt_boolean(text, title="TiLDA", true_text="Yes", false_text="No", width
     if style == None:
         style = default_style_dialog
     ugfx.set_default_font(FONT_MEDIUM_BOLD)
-    window = ugfx.Container((ugfx.width() - width) // 2, (ugfx.height() - height) // 2,  width, height, style=style)
+    window = ugfx.Container((ugfx.width() - width) // 2, (ugfx.height() - height) // 2,  width, height)
     window.show()
     ugfx.set_default_font(font)
     window.text(5, 10, title, TILDA_COLOR)
@@ -49,17 +49,15 @@ def prompt_boolean(text, title="TiLDA", true_text="Yes", false_text="No", width
     button_no = ugfx.Button(width // 2 + 5, height - 40, width // 2 - 15, 30 , false_text, parent=window) if false_text else None
 
     try:
-        buttons.init()
+        #button_yes.attach_input(ugfx.BTN_A,0) # todo: re-enable once working
-
+        #if button_no: button_no.attach_input(ugfx.BTN_B,0)
-        button_yes.attach_input(ugfx.BTN_A,0)
-        if button_no: button_no.attach_input(ugfx.BTN_B,0)
 
         window.show()
 
         while True:
             sleep.wfi()
-            if buttons.is_triggered("BTN_A"): return True
+            if buttons.is_triggered(buttons.Buttons.BTN_A): return True
-            if buttons.is_triggered("BTN_B"): return False
+            if buttons.is_triggered(buttons.Buttons.BTN_B): return False
 
     finally:
         window.hide()
@@ -68,13 +66,13 @@ def prompt_boolean(text, title="TiLDA", true_text="Yes", false_text="No", width
         if button_no: button_no.destroy()
         label.destroy()
 
-def prompt_text(description, init_text = "", true_text="OK", false_text="Back", width = 300, height = 200, font=FONT_MEDIUM_BOLD, style=default_style_badge):
+def prompt_text(description, init_text = "", true_text="OK", false_text="Back", font=FONT_MEDIUM_BOLD, style=default_style_badge):
     """Shows a dialog and keyboard that allows the user to input/change a string
 
     Returns None if user aborts with button B
     """
 
-    window = ugfx.Container(int((ugfx.width()-width)/2), int((ugfx.height()-height)/2), width, height, style=style)
+    window = ugfx.Container(0, 0, ugfx.width(), ugfx.height())
 
     if false_text:
         true_text = "M: " + true_text
@@ -83,29 +81,27 @@ def prompt_text(description, init_text = "", true_text="OK", false_text="Back",
     if buttons.has_interrupt("BTN_MENU"):
         buttons.disable_interrupt("BTN_MENU")
 
-    ugfx.set_default_font(ugfx.FONT_MEDIUM)
+    ugfx.set_default_font(FONT_MEDIUM_BOLD)
-    kb = ugfx.Keyboard(0, int(height/2), width, int(height/2), parent=window)
+    kb = ugfx.Keyboard(0, ugfx.height()//2, ugfx.width(), ugfx.height()//2, parent=window)
-    edit = ugfx.Textbox(5, int(height/2)-30, int(width*4/5)-10, 25, text = init_text, parent=window)
+    edit = ugfx.Textbox(2, ugfx.height()//2-60, ugfx.width()-7, 25, text = init_text, parent=window)
     ugfx.set_default_font(FONT_SMALL)
-    button_yes = ugfx.Button(int(width*4/5), int(height/2)-30, int(width*1/5)-3, 25 , true_text, parent=window)
+    button_yes = ugfx.Button(2, ugfx.height()//2-30, ugfx.width()//2-6, 25 , true_text, parent=window)
-    button_no = ugfx.Button(int(width*4/5), int(height/2)-30-30, int(width/5)-3, 25 , false_text, parent=window) if false_text else None
+    button_no = ugfx.Button(ugfx.width()//2+2, ugfx.height()//2-30, ugfx.width()//2-6, 25 , false_text, parent=window) if false_text else None
     ugfx.set_default_font(font)
-    label = ugfx.Label(int(width/10), int(height/10), int(width*4/5), int(height*2/5)-60, description, parent=window)
+    label = ugfx.Label(ugfx.width()//10, ugfx.height()//10, ugfx.width()*4//5, ugfx.height()*2//5-90, description, parent=window)
 
     try:
-        buttons.init()
+        #button_yes.attach_input(ugfx.BTN_MENU,0) # todo: re-enable this
-
+        #if button_no: button_no.attach_input(ugfx.BTN_B,0)
-        button_yes.attach_input(ugfx.BTN_MENU,0)
-        if button_no: button_no.attach_input(ugfx.BTN_B,0)
 
         window.show()
-        edit.set_focus()
+        # edit.set_focus() todo: do we need this?
         while True:
             sleep.wfi()
             ugfx.poll()
-            #if buttons.is_triggered("BTN_A"): return edit.text()
+            #if buttons.is_triggered(buttons.Buttons.BTN_A): return edit.text()
-            if buttons.is_triggered("BTN_B"): return None
+            if buttons.is_triggered(buttons.Buttons.BTN_B): return None
-            if buttons.is_triggered("BTN_MENU"): return edit.text()
+            if buttons.is_triggered(buttons.Buttons.BTN_Menu): return edit.text()
 
     finally:
         window.hide()
@@ -143,7 +139,7 @@ def prompt_option(options, index=0, text = "Please select one of the following:"
             options_list.add_item(option["title"])
         else:
             options_list.add_item(str(option))
-    options_list.selected_index(index)
+    options_list.set_selected_index(index)
 
     select_text = "A: " + select_text
     if none_text:
@@ -153,14 +149,20 @@ def prompt_option(options, index=0, text = "Please select one of the following:"
     button_none = ugfx.Button(ugfx.width() - 160, ugfx.height() - 50, 140, 30 , none_text, parent=window) if none_text else None
 
     try:
-        buttons.init()
-
         while True:
             sleep.wfi()
             ugfx.poll()
-            if buttons.is_triggered("BTN_A"): return options[options_list.selected_index()]
+            # todo: temporary hack
-            if button_none and buttons.is_triggered("BTN_B"): return None
+            if (buttons.is_triggered(buttons.Buttons.JOY_Up)):
-            if button_none and buttons.is_triggered("BTN_MENU"): return None
+                index = max(index - 1, 0)
+                options_list.set_selected_index(index)
+            if (buttons.is_triggered(buttons.Buttons.JOY_Down)):
+                index = min(index + 1, len(options) - 1)
+                options_list.set_selected_index(index)
+
+            if buttons.is_triggered(buttons.Buttons.BTN_A): return options[options_list.get_selected_index()]
+            if button_none and buttons.is_triggered(buttons.Buttons.BTN_B): return None
+            if button_none and buttons.is_triggered(buttons.Buttons.BTN_Menu): return None
 
     finally:
         window.hide()

lib/http.py

@@ -168,11 +168,7 @@ def open_http_socket(method, url, json=None, timeout=None, headers=None, data=No
         content = None
 
     # ToDo: Handle IPv6 addresses
-    if is_ipv4_address(host):
+    addr = get_address_info(host, port)
-        addr = (host, port)
-    else:
-        ai = usocket.getaddrinfo(host, port)
-        addr = ai[0][4]
 
     sock = None
     if proto == 'https:':
@@ -203,6 +199,21 @@ def open_http_socket(method, url, json=None, timeout=None, headers=None, data=No
 
     return sock
 
+def get_address_info(host, port, retries_left = 20):
+    try:
+        if is_ipv4_address(host):
+            addr = (host, port)
+        else:
+            return usocket.getaddrinfo(host, port)[0][4]
+    except OSError as e:
+        if ("-15" in str(e)) and retries_left:
+            # [addrinfo error -15]
+            # This tends to happen after startup and goes away after a while
+            time.sleep_ms(200)
+            return get_address_info(host, port, retries_left - 1)
+        else:
+            raise e
+
 # Adapted from upip
 def request(method, url, json=None, timeout=None, headers=None, data=None, params=None):
     sock = open_http_socket(method, url, json, timeout, headers, data, params)

lib/test_dialogs.py

@@ -3,9 +3,9 @@
 Very limited at the moment since we can't test the main input dialogs"""
 
 ___license___      = "MIT"
-___dependencies___ = ["upip:unittest", "dialogs", "sleep"]
+___dependencies___ = ["upip:unittest", "dialogs", "sleep", "ugfx_helper"]
 
-import unittest, ugfx
+import unittest, ugfx, ugfx_helper
 from machine import Pin
 from dialogs import *
 from sleep import *
@@ -13,20 +13,21 @@ from sleep import *
 class TestDialogs(unittest.TestCase):
 
     def setUpClass(self):
-        ugfx.init()
+        ugfx_helper.init()
-        Pin(Pin.PWM_LCD_BLIGHT).on()
 
     def tearDownClass(self):
-        Pin(Pin.PWM_LCD_BLIGHT).off()
+        ugfx_helper.deinit()
 
-    def test_app_object(self):
+#    def test_waiting(self):
-        count_max = 10
+#        count_max = 3
-        with WaitingMessage("Testing...", "Foo") as c:
+#        with WaitingMessage("Testing...", "Foo") as c:
-            for i in range(1, count_max):
+#            for i in range(1, count_max):
-                sleep_ms(100)
+#                c.text = "%d/%d" % (i, count_max)
-                c.text = "%d/%d" % (i, count_max)
+#
+#        print("done")
 
-        print("done")
+    def test_(self):
+        prompt_text("description")
 
 
 

lib/test_hall_effect.py

@@ -3,7 +3,7 @@
 ___license___      = "MIT"
 ___dependencies___ = ["upip:unittest", "hall_effect"]
 
-import unittest, hall_effect
+import unittest, hall_effect, speaker
 
 class TestHallEffect(unittest.TestCase):
 
@@ -12,6 +12,5 @@ class TestHallEffect(unittest.TestCase):
         self.assertTrue(flux > 0)
         self.assertTrue(flux < 4000)
 
-
 if __name__ == '__main__':
     unittest.main()

lib/test_wifi.py

@@ -3,13 +3,21 @@
 ___license___      = "MIT"
 ___dependencies___ = ["upip:unittest", "wifi"]
 
-import unittest, wifi
+import unittest, wifi, ugfx
+from machine import Pin
 
 class TestWifi(unittest.TestCase):
 
+    def setUpClass(self):
+        ugfx.init()
+        Pin(Pin.PWM_LCD_BLIGHT).on()
+
+    def tearDownClass(self):
+        Pin(Pin.PWM_LCD_BLIGHT).off()
+
     def test_connect(self):
         wifi.connect()
-        self.assertTrue(wifi.is_connected())
+        self.assertTrue(wifi.is_connected(show_wait_message=True))
 
 if __name__ == '__main__':
     unittest.main()

lib/wifi.py

@@ -31,9 +31,7 @@ def ssid():
     return connection_details()["ssid"]
 
 def connect(wait=True, timeout=10, show_wait_message=False, prompt_on_fail=True, dialog_title='TiLDA'):
-    retry_connect = True
+    while True:
-
-    while retry_connect:
         if nic().isconnected():
             return
 
@@ -60,7 +58,7 @@ def connect(wait=True, timeout=10, show_wait_message=False, prompt_on_fail=True,
                         text="Failed to connect to '%s'" % details['ssid'],
                         title=dialog_title,
                         true_text="Try again",
-                        false_text="Forget it",
+                        false_text="Change it",
                     )
                     if not retry_connect:
                         os.remove('wifi.json')
@@ -76,11 +74,11 @@ def connect_wifi(details, timeout, wait=False):
         nic().connect(details['ssid'])
 
     if wait:
-        wait_until = time.ticks_ms() + 2000
+        wait_until = time.ticks_ms() + timeout * 1000
         while not nic().isconnected():
             #nic().update() # todo: do we need this?
             if (time.ticks_ms() > wait_until):
-                raise Exception("Timeout while trying to connect to wifi")
+                raise OSError("Timeout while trying to connect to wifi")
             sleep.sleep_ms(100)
 
 
@@ -88,6 +86,7 @@ def is_connected():
     return nic().isconnected()
 
 def get_security_level(ap):
+    #todo: fix this
     n = nic()
     levels = {}
     try:
@@ -106,22 +105,30 @@ def get_security_level(ap):
 def choose_wifi(dialog_title='TiLDA'):
     filtered_aps = []
     with dialogs.WaitingMessage(text='Scanning for networks...', title=dialog_title):
-        visible_aps = nic().list_aps()
+        visible_aps = None
-        visible_aps.sort(key=lambda x:x['rssi'], reverse=True)
+        while not visible_aps:
+            visible_aps = nic().scan()
+            print(visible_aps)
+            sleep.sleep_ms(300)
+            #todo: timeout
+        print(visible_aps)
+        visible_aps.sort(key=lambda x:x[3], reverse=True)
+        print(visible_aps)
         # We'll get one result for each AP, so filter dupes
         for ap in visible_aps:
-            title = ap['ssid']
+            title = ap[0]
-            security = get_security_level(ap)
+            security = "?" # todo: re-add get_security_level(ap)
             if security:
                 title = title + ' (%s)' % security
             ap = {
                 'title': title,
-                'ssid': ap['ssid'],
+                'ssid': ap[0],
-                'security': security,
+                'security': ap[4],
             }
             if ap['ssid'] not in [ a['ssid'] for a in filtered_aps ]:
                 filtered_aps.append(ap)
         del visible_aps
+    print(filtered_aps)
 
     ap = dialogs.prompt_option(
         filtered_aps,