Show COVID-19 exposure notification statistics

Based on !390 (merged)

Displays how many unique MACs (aka phones) which send exposure notifications have been seen during the last 100 seconds.

It tries to distinguish between Apple and Google devices based on a heuristic mentioned in https://github.com/znuh/microbit-corona-scanner. The user can choose if they want the LEDs and/or the vibration motor to show newly discovered MACs / exposure notifications.

Loosely based on:

• https://github.com/kauzu/CoronaAppDetector
• https://github.com/frankrieger/CoronaAppDetectorM5StickC
• https://github.com/znuh/microbit-corona-scanner

Most of the logic resides in a plain Python script and can be modified by the user as needed.

preload/apps/en_count/__init__.py

@@ -4,6 +4,8 @@ import time
 import vibra
 import display
 import color
+import buttons
+import leds
 
 DM_ADV_TYPE_FLAGS = 0x01
 DM_ADV_TYPE_16_UUID = 0x03
@@ -11,7 +13,14 @@ DM_ADV_TYPE_SERVICE_DATA = 0x16
 UUID = b"\x6f\xfd"
 TIMEOUT = 100
 
+MODE_OFF = 0
+MODE_ON_NEW_MAC = 1
+MODE_ON_RX = 2
+MODE_BOTH = 2
+
 seen = {}
+vib_mode = MODE_OFF
+led_mode = MODE_OFF
 
 
 def parse_advertisement_data(data):
@@ -37,12 +46,25 @@ def bytes2hex(bin, sep=""):
     return sep.join(["%02x" % x for x in bin])
 
 
-def process_covid_dada(mac, service_data, rssi, flags):
-    vibra.vibrate(10)
+def process_covid_data(mac, service_data, rssi, flags):
+    global vib_mode
+    if vib_mode in [MODE_ON_RX, MODE_BOTH]:
+        vibra.vibrate(10)
+
+    if vib_mode in [MODE_ON_NEW_MAC, MODE_BOTH] and mac not in seen:
+        vibra.vibrate(100)
+
+    if led_mode in [MODE_ON_RX, MODE_BOTH]:
+        leds.flash_rocket(0, 31, 20)
+
+    if led_mode in [MODE_ON_NEW_MAC, MODE_BOTH] and mac not in seen:
+        leds.flash_rocket(1, 31, 200)
+
     print(bytes2hex(mac, ":"), rssi, bytes2hex(service_data), flags)
 
     # try to produce a small int
-    seen[mac] = [int(time.time() - t0), flags]
+    last_rx_time = time.time() - t0
+    seen[mac] = [int(last_rx_time), flags]
 
 
 def prune():
@@ -73,7 +95,7 @@ def process_scan_report(scan_report):
                 if DM_ADV_TYPE_FLAGS in ads:
                     flags = ads[DM_ADV_TYPE_FLAGS]
                 # service data contains another copy of the service UUID
-                process_covid_dada(mac, ads[DM_ADV_TYPE_SERVICE_DATA][2:], rssi, flags)
+                process_covid_data(mac, ads[DM_ADV_TYPE_SERVICE_DATA][2:], rssi, flags)
 
 
 def ble_callback(_):
@@ -86,30 +108,28 @@ def ble_callback(_):
         process_scan_report(scan_report)
 
 
-t0 = time.time()
-disp = display.open()
-
-interrupt.set_callback(interrupt.BLE, ble_callback)
-interrupt.enable_callback(interrupt.BLE)
-
-sys_ble.scan_start()
-
-while True:
+def show_stats():
     seen_google = 0
     seen_apple = 0
+
     t = time.time() - t0
 
     t_min = t
-    for mac in seen:
-        if seen[mac][1]:
+
+    # Make a copy as `seen` could change in between
+    # and we're not locking it
+    seen_copy = seen.copy()
+    for mac in seen_copy:
+        info = seen_copy[mac]
+        if info[1]:
             seen_apple += 1
         else:
             seen_google += 1
-        if seen[mac][0] < t_min:
-            t_min = seen[mac][0]
+        if info[0] < t_min:
+            t_min = info[0]
     seen_total = seen_google + seen_apple
 
-    window = t - t_min
+    window = t - t_min if len(seen_copy) > 0 else min(TIMEOUT, t - last_rx_time)
 
     disp.clear()
     disp.print("Last %u s:" % window, posy=0, fg=color.WHITE)
@@ -117,6 +137,79 @@ while True:
     disp.print("Apple:  %d" % seen_apple, posy=40, fg=color.BLUE)
     disp.print("Total:  %d" % seen_total, posy=60, fg=color.WHITE)
     disp.update()
-    # print(seen)
 
-    time.sleep(1)
+
+t0 = time.time()
+last_rx_time = 0
+disp = display.open()
+v_old = 0
+pause = 1
+
+interrupt.set_callback(interrupt.BLE, ble_callback)
+interrupt.enable_callback(interrupt.BLE)
+
+
+disp.clear()
+disp.print(" Exp Notif", posy=0, fg=color.WHITE)
+disp.print("  Counter", posy=20, fg=color.WHITE)
+disp.print("Vib Mode ->", posy=40, fg=color.GREEN)
+disp.print("<- LED Mode", posy=60, fg=color.BLUE)
+disp.update()
+
+time.sleep(3)
+
+sys_ble.scan_start()
+
+while True:
+    v_new = buttons.read()
+    v = ~v_old & v_new
+    v_old = v_new
+
+    if v & buttons.TOP_RIGHT:
+        disp.clear()
+        disp.print("Vib Mode:", posy=0, fg=color.WHITE)
+        if vib_mode == MODE_OFF:
+            vib_mode = MODE_ON_NEW_MAC
+            disp.print("New MAC", posy=40, fg=color.YELLOW)
+        elif vib_mode == MODE_ON_NEW_MAC:
+            vib_mode = MODE_ON_RX
+            disp.print("Each RX", posy=40, fg=color.BLUE)
+        elif vib_mode == MODE_ON_RX:
+            vib_mode = MODE_BOTH
+            disp.print("MAC", posy=40, fg=color.YELLOW)
+            disp.print("&", posy=40, posx=14 * 4, fg=color.WHITE)
+            disp.print("RX", posy=40, posx=14 * 6, fg=color.BLUE)
+        elif vib_mode == MODE_BOTH:
+            vib_mode = MODE_OFF
+            disp.print("Vib off", posy=40, fg=color.WHITE)
+
+        disp.update()
+        pause = 20
+
+    if v & buttons.BOTTOM_LEFT:
+        disp.clear()
+        disp.print("LED Mode:", posy=0, fg=color.WHITE)
+        if led_mode == MODE_OFF:
+            led_mode = MODE_ON_NEW_MAC
+            disp.print("New MAC", posy=40, fg=color.YELLOW)
+        elif led_mode == MODE_ON_NEW_MAC:
+            led_mode = MODE_ON_RX
+            disp.print("Each RX", posy=40, fg=color.BLUE)
+        elif led_mode == MODE_ON_RX:
+            led_mode = MODE_BOTH
+            disp.print("MAC", posy=40, fg=color.YELLOW)
+            disp.print("&", posy=40, posx=14 * 4, fg=color.WHITE)
+            disp.print("RX", posy=40, posx=14 * 6, fg=color.BLUE)
+        elif led_mode == MODE_BOTH:
+            led_mode = MODE_OFF
+            disp.print("LED off", posy=40, fg=color.WHITE)
+
+        disp.update()
+        pause = 20
+
+    pause -= 1
+    if pause == 0:
+        show_stats()
+        pause = 10
+
+    time.sleep(0.1)