v3 Add NEC_IR. Tutorial corrections and improvements.

Author: peterhinch

v3/README.md

@@ -21,8 +21,12 @@ Documented in the tutorial.
 Documented in the tutorial.
 
 #### Asynchronous device drivers  
-The device drivers are in the process of being ported. Currently there is a
-[GPS driver](./docs/GPS.md).
+
+The device drivers are in the process of being ported. These currently
+comprise:
+
+ * [GPS driver](./docs/GPS.md) Includes various GPS utilities.
+ * [HTU21D](./docs/HTU21D.md) Temperature and humidity sensor.
 
 # 2 V3 Overview
 

v3/as_demos/__init__.py

@@ -17,14 +17,21 @@ async def toggle(objLED, time_ms):
 
 # TEST FUNCTION
 
-def test(duration):
-    duration = int(duration)
-    if duration > 0:
-        print("Flash LED's for {:3d} seconds".format(duration))
+async def main(duration):
+    print("Flash LED's for {} seconds".format(duration))
     leds = [pyb.LED(x) for x in range(1,4)]  # Initialise three on board LED's
-    for x, led in enumerate(leds):           # Create a coroutine for each LED
+    for x, led in enumerate(leds):  # Create a task for each LED
         t = int((0.2 + x/2) * 1000)
         asyncio.create_task(toggle(leds[x], t))
     asyncio.run(killer(duration))
 
-test(10)
+def test(duration=10):
+    try:
+        asyncio.run(main(duration))
+    except KeyboardInterrupt:
+        print('Interrupted')
+    finally:
+        asyncio.new_event_loop()
+        print('as_demos.aledflash.test() to run again.')
+
+test()

v3/as_demos/aledflash.py

@@ -25,4 +25,13 @@ async def main():
     while True:
         await asyncio.sleep(1)
 
-asyncio.run(main())
+def test():
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        print('Interrupted')
+    finally:
+        asyncio.new_event_loop()
+        print('as_demos.auart.test() to run again.')
+
+test()

v3/as_demos/auart.py

@@ -68,7 +68,7 @@ async def send_command(self, command):
             await asyncio.sleep(1)  # Wait for 4s after last msg received
         return self.response
 
-async def test():
+async def main():
     print('This test takes 10s to complete.')
     master = Master()
     device = Device()
@@ -101,6 +101,14 @@ def printexp():
     print(st)
     print('\x1b[39m')
 
-printexp()
-asyncio.run(test())
+def test():
+    printexp()
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        print('Interrupted')
+    finally:
+        asyncio.new_event_loop()
+        print('as_demos.auart_hd.test() to run again.')
 
+test()

v3/as_demos/auart_hd.py

@@ -18,8 +18,9 @@ async def foo(n):
         while True:
             await asyncio.sleep(1)
             n += 1
-    except Exception as e: #asyncio.TimeoutError:
-        print('foo timeout.', e)
+    except asyncio.CancelledError:
+        print('Trapped foo timeout.')
+        raise
     return n
 
 async def bar(n):
@@ -28,22 +29,72 @@ async def bar(n):
         while True:
             await asyncio.sleep(1)
             n += 1
-    except Exception as e:
-        print('bar stopped.', e)
+    except asyncio.CancelledError:  # Demo of trapping
+        print('Trapped bar cancellation.')
+        raise
     return n
 
 async def do_cancel(task):
     await asyncio.sleep(5)
     print('About to cancel bar')
     task.cancel()
 
-async def main():
+async def main(rex):
     bar_task = asyncio.create_task(bar(70))  # Note args here
     tasks = []
     tasks.append(barking(21))
     tasks.append(asyncio.wait_for(foo(10), 7))
     asyncio.create_task(do_cancel(bar_task))
-    res = await asyncio.gather(*tasks)
+    try:
+        res = await asyncio.gather(*tasks, return_exceptions=rex)
+    except asyncio.TimeoutError:
+        print('foo timed out.')
+        res = 'No result'
     print('Result: ', res)
 
-asyncio.run(main())
+
+exp_false = '''Test runs for 10s. Expected output:
+
+Start cancellable bar()
+Start normal coro barking()
+Start timeout coro foo()
+About to cancel bar
+Trapped bar cancellation.
+Done barking.
+Trapped foo timeout.
+foo timed out.
+Result:  No result
+
+'''
+exp_true = '''Test runs for 10s. Expected output:
+
+Start cancellable bar()
+Start normal coro barking()
+Start timeout coro foo()
+About to cancel bar
+Trapped bar cancellation.
+Done barking.
+Trapped foo timeout.
+Result:  [42, TimeoutError()]
+
+'''
+
+def printexp(st):
+    print('\x1b[32m')
+    print(st)
+    print('\x1b[39m')
+
+def test(rex):
+    st = exp_true if rex else exp_false
+    printexp(st)
+    try:
+        asyncio.run(main(rex))
+    except KeyboardInterrupt:
+        print('Interrupted')
+    finally:
+        asyncio.new_event_loop()
+        print()
+        print('as_demos.gather.test() to run again.')
+        print('as_demos.gather.test(True) to see effect of return_exceptions.')
+
+test(rex=False)

v3/as_demos/gather.py

@@ -1 +1 @@
-from .as_GPS import *
+from .aremote import *

v3/as_drivers/nec_ir/__init__.py

@@ -6,10 +6,10 @@
 
 from sys import platform
 import uasyncio as asyncio
-from asyn import Event
+from primitives.message import Message
 from micropython import const
 from array import array
-from utime import ticks_us, ticks_diff
+from utime import ticks_ms, ticks_us, ticks_diff
 if platform == 'pyboard':
     from pyb import Pin, ExtInt
 else:
@@ -41,7 +41,7 @@
 # Value of 73 allows for up to 35ms latency.
 class NEC_IR():
     def __init__(self, pin, callback, extended, *args):  # Optional args for callback
-        self._ev_start = Event()
+        self._ev_start = Message()
         self._callback = callback
         self._extended = extended
         self._addr = 0
@@ -56,15 +56,13 @@ def __init__(self, pin, callback, extended, *args):  # Optional args for callbac
             pin.irq(handler = self._cb_pin, trigger = (Pin.IRQ_FALLING | Pin.IRQ_RISING), hard = True)
         self._edge = 0
         self._ev_start.clear()
-        loop = asyncio.get_event_loop()
-        loop.create_task(self._run())
+        asyncio.create_task(self._run())
 
     async def _run(self):
-        loop = asyncio.get_event_loop()
         while True:
             await self._ev_start  # Wait until data collection has started
             # Compensate for asyncio latency
-            latency = ticks_diff(loop.time(), self._ev_start.value())
+            latency = ticks_diff(ticks_ms(), self._ev_start.value())
             await asyncio.sleep_ms(self.block_time - latency)  # Data block should have ended
             self._decode()  # decode, clear event, prepare for new rx, call cb
 
@@ -74,8 +72,7 @@ def _cb_pin(self, line):
         # On overrun ignore pulses until software timer times out
         if self._edge <= _EDGECOUNT:  # Allow 1 extra pulse to record overrun
             if not self._ev_start.is_set():  # First edge received
-                loop = asyncio.get_event_loop()
-                self._ev_start.set(loop.time())  # asyncio latency compensation
+                self._ev_start.set(ticks_ms())  # asyncio latency compensation
             self._times[self._edge] = t
             self._edge += 1
 

v3/as_drivers/nec_ir/aremote.py

@@ -5,6 +5,7 @@
 # Copyright Peter Hinch 2017 Released under the MIT license
 
 # Run this to characterise a remote.
+# import as_drivers.nec_ir.art
 
 from sys import platform
 import uasyncio as asyncio
@@ -17,7 +18,7 @@
 else:
     print('Unsupported platform', platform)
 
-from aremote import *
+from .aremote import *
 
 errors = {BADSTART : 'Invalid start pulse', BADBLOCK : 'Error: bad block',
           BADREP : 'Error: repeat', OVERRUN : 'Error: overrun',
@@ -33,6 +34,7 @@ def cb(data, addr):
 
 def test():
     print('Test for IR receiver. Assumes NEC protocol.')
+    print('ctrl-c to stop.')
     if platform == 'pyboard':
         p = Pin('X3', Pin.IN)
     elif platform == 'esp8266':
@@ -42,6 +44,11 @@ def test():
         p = Pin(23, Pin.IN)
     ir = NEC_IR(p, cb, True)  # Assume r/c uses extended addressing
     loop = asyncio.get_event_loop()
-    loop.run_forever()
+    try:
+        loop.run_forever()
+    except KeyboardInterrupt:
+        print('Interrupted')
+    finally:
+        asyncio.new_event_loop()  # Still need ctrl-d because of interrupt vector
 
 test()

v3/as_drivers/nec_ir/art.py

@@ -55,6 +55,11 @@ def test():
         led(1)
     ir = NEC_IR(p, cb, True, led)  # Assume extended address mode r/c
     loop = asyncio.get_event_loop()
-    loop.run_forever()
+    try:
+        loop.run_forever()
+    except KeyboardInterrupt:
+        print('Interrupted')
+    finally:
+        asyncio.new_event_loop()  # Still need ctrl-d because of interrupt vector
 
 test()

v3/as_drivers/nec_ir/art1.py

@@ -8,17 +8,32 @@ microcontroller.
 
 The driver and test programs run on the Pyboard and ESP8266.
 
-# Files
+## An alternative solution
 
- 1. `aremote.py` The device driver.
- 2. `art.py` A test program to characterise a remote.
- 3. `art1.py` Control an onboard LED using a remote. The data and addresss
- values need changing to match your characterised remote.
+This solution provides an example of an asynchronous device driver. A more
+complete IR solution may be found
+[here](https://github.com/peterhinch/micropython_ir). This supports other
+protocols and IR "blasting". It does not use `uasyncio` but is nonblocking and
+is compatible with `uasyncio` applications.
+
+# Demo scripts
+
+The following prints data and address values received from a remote. These
+values enable you to respond to individual butons.
+```python
+import as_drivers.nec_ir.art
+```
+
+Control an onboard LED using a remote. The data and addresss values must be
+changed to match your characterised remote.
+```python
+import as_drivers.nec_ir.art1
+```
 
 # Dependencies
 
-The driver requires the `uasyncio` library and the file `asyn.py` from this
-repository.
+The driver requires the `uasyncio` library and the `primitives` package from
+this repository.
 
 # Usage
 
@@ -131,3 +146,9 @@ owing to high interrupt latency.
 `BADDATA` Data did not match check byte.  
 `BADADDR` Where `extended` is `False` the 8-bit address is checked
 against the check byte. This code is returned on failure.  
+
+# Files
+
+ 1. `aremote.py` The device driver.
+ 2. `art.py` A test program to characterise a remote.
+ 3. `art1.py`