Minor code improvements, notably to demos.

Author: peterhinch

README.md

@@ -636,7 +636,7 @@ This will pause until connectivity has been established. It can be issued at
 any time: if the `Connection` has already been instantiated, that instance will
 be returned. The `Connection` constructor should not be called by applications.
 
-The `Connection` instance:
+#### The `Connection` instance
 
 Methods (asynchronous):
  1. `readline` No args. Pauses until data received. Returns a line.
@@ -656,6 +656,10 @@ Methods (synchronous):
 Class Method (synchronous):
  1. `close_all` No args. Closes all sockets: call on exception (e.g. ctrl-c).
 
+Bound variable:
+ 1. `nconns` Maintains a count of (re)connections for information or monitoring
+ of outages.
+
 The `Connection` class is awaitable. If
 ```python
 await connection_instance

iot/__init__.py

@@ -0,0 +1,46 @@
+# __init__.py Common utility functions for micropython-iot
+
+# Released under the MIT licence.
+# Copyright (C) Peter Hinch 2019-2020
+
+# Now uses and requires uasyncio V3. This is incorporated in daily builds
+# and release builds later than V1.12
+# Under CPython requires CPython 3.8 or later.
+
+# Create message ID's. Initially 0 then 1 2 ... 254 255 1 2
+def gmid():
+    mid = 0
+    while True:
+        yield mid
+        mid = (mid + 1) & 0xff
+        mid = mid if mid else 1
+
+# Return True if a message ID has not already been received
+def isnew(mid, lst=bytearray(32)):
+    if mid == -1:
+        for idx in range(32):
+            lst[idx] = 0
+        return
+    idx = mid >> 3
+    bit = 1 << (mid & 7)
+    res = not(lst[idx] & bit)
+    lst[idx] |= bit
+    lst[(idx + 16 & 0x1f)] = 0
+    return res
+
+# Minimal implementation of set for integers in range 0-255
+class SetByte:
+    def __init__(self):
+        self._ba = bytearray(32)
+
+    def __bool__(self):
+        return any(self._ba)
+
+    def __contains__(self, i):
+        return (self._ba[i >> 3] & 1 << (i & 7)) > 0
+
+    def discard(self, i):
+        self._ba[i >> 3] &= ~(1 << (i &7))
+
+    def add(self, i):
+        self._ba[i >> 3] |= 1 << (i & 7)

iot/client.mpy

@@ -22,10 +22,8 @@
 import utime
 import machine
 import uerrno as errno
-from .primitives import gmid, isnew, launch, SetByte  # __init__.py
-Event = asyncio.Event
-Lock = asyncio.Lock
-
+from . import gmid, isnew, SetByte  # __init__.py
+from .primitives import launch
 gc.collect()
 from micropython import const
 
@@ -103,8 +101,8 @@ def inner(feed=WDT_CB):
             import esp
             esp.sleep_type(esp.SLEEP_NONE)  # Improve connection integrity at cost of power consumption.
 
-        self._evfail = Event()
-        self._s_lock = Lock()  # For internal send conflict.
+        self._evfail = asyncio.Event()
+        self._s_lock = asyncio.Lock()  # For internal send conflict.
         self._last_wr = utime.ticks_ms()
         self._lineq = deque((), 20, True)  # 20 entries, throw on overflow
         self.connects = 0  # Connect count for test purposes/app access
@@ -185,9 +183,8 @@ async def _write(self, line):
             # After an outage wait until something is received from server
             # before we send.
             await self
-            async with self._s_lock:
-                if await self._send(line):
-                    return
+            if await self._send(line):
+                return
 
             # send fail. _send has triggered _evfail. Await response.
             while self():
@@ -263,14 +260,7 @@ async def _run(self):
                 tsk_reader = asyncio.create_task(self._reader())
                 # Server reads ID immediately, but a brief pause is probably wise.
                 await asyncio.sleep_ms(50)
-                # No need for lock yet.
-                try:
-                    if not await self._send(self._my_id):
-                        raise OSError
-                except OSError:
-                    if init:
-                        await self.bad_server()
-                else:
+                if await self._send(self._my_id):
                     tsk_ka = asyncio.create_task(self._keepalive())
                     if self._concb is not None:
                         # apps might need to know connection to the server acquired
@@ -284,6 +274,8 @@ async def _run(self):
                     if self._concb is not None:
                         # apps might need to know if they lost connection to the server
                         launch(self._concb, False, *self._concbargs)
+                elif init:
+                    await self.bad_server()
             finally:
                 init = False
                 self._close()  # Close socket but not wdt
@@ -325,16 +317,14 @@ async def _reader(self):  # Entry point is after a (re) connect.
                 self.connects += 1  # update connect count
 
     async def _sendack(self, mid):
-        async with self._s_lock:
-            await self._send('{:02x}\n'.format(mid))
+        await self._send('{:02x}\n'.format(mid))
 
     async def _keepalive(self):
         while True:
             due = self._tim_ka - utime.ticks_diff(utime.ticks_ms(), self._last_wr)
             if due <= 0:
-                async with self._s_lock:
-                    # error sets ._evfail, .run cancels this coro
-                    await self._send(b'\n')
+                # error sets ._evfail, .run cancels this coro
+                await self._send(b'\n')
             else:
                 await asyncio.sleep_ms(due)
 
@@ -376,26 +366,27 @@ async def _readline(self, to):
                 line = b''.join((line, d)) if line else d
 
     async def _send(self, d):  # Write a line to socket.
-        start = utime.ticks_ms()
-        while d:
-            try:
-                ns = self._sock.send(d)  # OSError if client closes socket
-            except OSError as e:
-                err = e.args[0]
-                if err == errno.EAGAIN:  # Would block: await server read
-                    await asyncio.sleep_ms(100)
+        async with self._s_lock:
+            start = utime.ticks_ms()
+            while d:
+                try:
+                    ns = self._sock.send(d)  # OSError if client closes socket
+                except OSError as e:
+                    err = e.args[0]
+                    if err == errno.EAGAIN:  # Would block: await server read
+                        await asyncio.sleep_ms(100)
+                    else:
+                        self._verbose and print('_send fail. Disconnect')
+                        self._evfail.set()
+                        return False  # peer disconnect
                 else:
-                    self._verbose and print('_send fail. Disconnect')
-                    self._evfail.set()
-                    return False  # peer disconnect
-            else:
-                d = d[ns:]
-                if d:  # Partial write: pause
-                    await asyncio.sleep_ms(20)
-                if utime.ticks_diff(utime.ticks_ms(), start) > self._to:
-                    self._verbose and print('_send fail. Timeout.')
-                    self._evfail.set()
-                    return False
-
-        self._last_wr = utime.ticks_ms()
+                    d = d[ns:]
+                    if d:  # Partial write: pause
+                        await asyncio.sleep_ms(20)
+                    if utime.ticks_diff(utime.ticks_ms(), start) > self._to:
+                        self._verbose and print('_send fail. Timeout.')
+                        self._evfail.set()
+                        return False
+
+            self._last_wr = utime.ticks_ms()
         return True

iot/client.py

@@ -1,12 +1,8 @@
-# __init__.py Common utility functions for micropython-iot
+# __init__.py Common functions for uasyncio primitives
 
 # Released under the MIT licence.
 # Copyright (C) Peter Hinch 2019-2020
 
-# Now uses and requires uasyncio V3. This is incorporated in daily builds
-# and release builds later than V1.12
-# Under CPython requires CPython 3.8 or later.
-
 try:
     import uasyncio as asyncio
 except ImportError:
@@ -23,44 +19,10 @@ def launch(func, *tup_args):
         loop = asyncio.get_event_loop()
         loop.create_task(res)
 
-
-# Create message ID's. Initially 0 then 1 2 ... 254 255 1 2
-def gmid():
-    mid = 0
-    while True:
-        yield mid
-        mid = (mid + 1) & 0xff
-        mid = mid if mid else 1
-
-# Return True if a message ID has not already been received
-def isnew(mid, lst=bytearray(32)):
-    if mid == -1:
-        for idx in range(32):
-            lst[idx] = 0
-        return
-    idx = mid >> 3
-    bit = 1 << (mid & 7)
-    res = not(lst[idx] & bit)
-    lst[idx] |= bit
-    lst[(idx + 16 & 0x1f)] = 0
-    return res
-
-# Minimal implementation of set for integers in range 0-255
-class SetByte:
-    def __init__(self):
-        self._ba = bytearray(32)
-
-    def __bool__(self):
-        for x in self._ba:
-            if x:
-                return True
-        return False
-
-    def __contains__(self, i):
-        return (self._ba[i >> 3] & 1 << (i & 7)) > 0
-
-    def discard(self, i):
-        self._ba[i >> 3] &= ~(1 << (i &7))
-
-    def add(self, i):
-        self._ba[i >> 3] |= 1 << (i & 7)
+def set_global_exception():
+    def _handle_exception(loop, context):
+        import sys
+        sys.print_exception(context["exception"])
+        sys.exit()
+    loop = asyncio.get_event_loop()
+    loop.set_exception_handler(_handle_exception)

iot/primitives/__init__.py

@@ -16,9 +16,10 @@
 gc.collect()
 from iot import client
 import urandom
+from .check_mid import CheckMid
 
 # Optional LED. led=None if not required
-from sys import platform
+from sys import platform, exit, print_exception
 if platform == 'pyboard':  # D series
     from pyb import LED
     led = LED(1)
@@ -27,17 +28,18 @@
     led = Pin(2, Pin.OUT, value=1)  # Optional LED
 # End of optionalLED
 
+def _handle_exception(loop, context):
+    print_exception(context["exception"])
+    exit()
+
 class App:
     def __init__(self, verbose):
         self.verbose = verbose
         self.cl = client.Client(local.MY_ID, local.SERVER,
                                 local.PORT, local.SSID, local.PW,
                                 local.TIMEOUT, verbose=verbose, led=led)
         self.tx_msg_id = 0
-        self.dupes = 0  # Incoming dupe count
-        self.missing = 0
-        self.last = 0
-        self.rxbuf = []
+        self.cm = CheckMid()  # Check message ID's for dupes, missing etc.
 
     async def start(self):
         self.verbose and print('App awaiting connection.')
@@ -51,30 +53,16 @@ async def reader(self):
         while True:
             line = await self.cl.readline()
             data = ujson.loads(line)
-            rxmid = data[0]
-            if rxmid in self.rxbuf:
-                self.dupes += 1
-            else:
-                self.rxbuf.append(rxmid)
+            self.cm(data[0])  # Update statistics
             print('Got', data, 'from server app')
 
-    def count_missed(self):
-        if len(self.rxbuf) >= 25:
-            idx = 0
-            while self.rxbuf[idx] < self.last + 10:
-                idx += 1
-            self.last += 10
-            self.missing += 10 - idx
-            self.rxbuf = self.rxbuf[idx:]
-        return self.missing
-
     # Send [ID, (re)connect count, free RAM, duplicate message count, missed msgcount]
     async def writer(self):
         self.verbose and print('Started writer')
         while True:
             gc.collect()
             data = [self.tx_msg_id, self.cl.connects, gc.mem_free(),
-                    self.dupes, self.count_missed()]
+                    self.cm.dupe, self.cm.miss]
             self.tx_msg_id += 1
             print('Sent', data, 'to server app\n')
             dstr = ujson.dumps(data)
@@ -88,6 +76,8 @@ def close(self):
 app = None
 async def main():
     global app  # For finally clause
+    loop = asyncio.get_event_loop()
+    loop.set_exception_handler(_handle_exception)
     app = App(verbose=True)
     await app.start()