Minor changes to switch and pushbutton drivers.

Author: peterhinch

v3/primitives/pushbutton.py

@@ -6,17 +6,20 @@
 import uasyncio as asyncio
 import utime as time
 from . import launch, Delay_ms
+
 try:
     from machine import TouchPad
 except ImportError:
     pass
 
+
 class Pushbutton:
     debounce_ms = 50
     long_press_ms = 1000
     double_click_ms = 400
+
     def __init__(self, pin, suppress=False, sense=None):
-        self.pin = pin # Initialise for input
+        self._pin = pin  # Initialise for input
         self._supp = suppress
         self._dblpend = False  # Doubleclick waiting for 2nd click
         self._dblran = False  # Doubleclick executed user function
@@ -25,10 +28,59 @@ def __init__(self, pin, suppress=False, sense=None):
         self._df = False
         self._ld = False  # Delay_ms instance for long press
         self._dd = False  # Ditto for doubleclick
-        self.sense = pin.value() if sense is None else sense  # Convert from electrical to logical value
-        self.state = self.rawstate()  # Initial state
-        self._run = asyncio.create_task(self.buttoncheck())  # Thread runs forever
+        # Convert from electrical to logical value
+        self._sense = pin.value() if sense is None else sense
+        self._state = self.rawstate()  # Initial state
+        self._run = asyncio.create_task(self._go())  # Thread runs forever
+
+    async def _go(self):
+        while True:
+            self._check(self.rawstate())
+            # Ignore state changes until switch has settled. Also avoid hogging CPU.
+            # See https://github.com/peterhinch/micropython-async/issues/69
+            await asyncio.sleep_ms(Pushbutton.debounce_ms)
+
+    def _check(self, state):
+        if state == self._state:
+            return
+        # State has changed: act on it now.
+        self._state = state
+        if state:  # Button pressed: launch pressed func
+            if self._tf:
+                launch(self._tf, self._ta)
+            if self._ld:  # There's a long func: start long press delay
+                self._ld.trigger(Pushbutton.long_press_ms)
+            if self._df:
+                if self._dd():  # Second click: timer running
+                    self._dd.stop()
+                    self._dblpend = False
+                    self._dblran = True  # Prevent suppressed launch on release
+                    launch(self._df, self._da)
+                else:
+                    # First click: start doubleclick timer
+                    self._dd.trigger(Pushbutton.double_click_ms)
+                    self._dblpend = True  # Prevent suppressed launch on release
+        else:  # Button release. Is there a release func?
+            if self._ff:
+                if self._supp:
+                    d = self._ld
+                    # If long delay exists, is running and doubleclick status is OK
+                    if not self._dblpend and not self._dblran:
+                        if (d and d()) or not d:
+                            launch(self._ff, self._fa)
+                else:
+                    launch(self._ff, self._fa)
+            if self._ld:
+                self._ld.stop()  # Avoid interpreting a second click as a long push
+            self._dblran = False
+
+    def _ddto(self):  # Doubleclick timeout: no doubleclick occurred
+        self._dblpend = False
+        if self._supp and not self._state:
+            if not self._ld or (self._ld and not self._ld()):
+                launch(self._ff, self._fa)
 
+    # ****** API ******
     def press_func(self, func=False, args=()):
         if func is None:
             self.press = asyncio.Event()
@@ -67,62 +119,19 @@ def long_func(self, func=False, args=()):
 
     # Current non-debounced logical button state: True == pressed
     def rawstate(self):
-        return bool(self.pin.value() ^ self.sense)
+        return bool(self._pin() ^ self._sense)
 
     # Current debounced state of button (True == pressed)
     def __call__(self):
-        return self.state
-
-    def _ddto(self):  # Doubleclick timeout: no doubleclick occurred
-        self._dblpend = False
-        if self._supp and not self.state:
-            if not self._ld or (self._ld and not self._ld()):
-                launch(self._ff, self._fa)
-
-    async def buttoncheck(self):
-        while True:
-            state = self.rawstate()
-            # State has changed: act on it now.
-            if state != self.state:
-                self.state = state
-                if state:  # Button pressed: launch pressed func
-                    if self._tf:
-                        launch(self._tf, self._ta)
-                    if self._ld:  # There's a long func: start long press delay
-                        self._ld.trigger(Pushbutton.long_press_ms)
-                    if self._df:
-                        if self._dd():  # Second click: timer running
-                            self._dd.stop()
-                            self._dblpend = False
-                            self._dblran = True  # Prevent suppressed launch on release
-                            launch(self._df, self._da)
-                        else:
-                            # First click: start doubleclick timer
-                            self._dd.trigger(Pushbutton.double_click_ms)
-                            self._dblpend = True  # Prevent suppressed launch on release
-                else:  # Button release. Is there a release func?
-                    if self._ff:
-                        if self._supp:
-                            d = self._ld 
-                            # If long delay exists, is running and doubleclick status is OK
-                            if not self._dblpend and not self._dblran:
-                                if (d and d()) or not d:
-                                    launch(self._ff, self._fa)
-                        else:
-                            launch(self._ff, self._fa)
-                    if self._ld:
-                        self._ld.stop()  # Avoid interpreting a second click as a long push
-                    self._dblran = False
-            # Ignore state changes until switch has settled
-            # See https://github.com/peterhinch/micropython-async/issues/69
-            await asyncio.sleep_ms(Pushbutton.debounce_ms)
+        return self._state
 
     def deinit(self):
         self._run.cancel()
 
 
 class ESP32Touch(Pushbutton):
     thresh = (80 << 8) // 100
+
     @classmethod
     def threshold(cls, val):
         if not (isinstance(val, int) and 0 < val < 100):

v3/primitives/tests/switches.py

@@ -21,12 +21,16 @@
 
 '''
 tests = '''
+\x1b[32m
 Available tests:
-test_sw Switch test
-test_swcb Switch with callback
-test_btn Pushutton launching coros
-test_btncb Pushbutton launching callbacks
+test_sw Switch test.
+test_swcb Switch with callback.
+test_sw_event Switch with event.
+test_btn Pushutton launching coros.
+test_btncb Pushbutton launching callbacks.
 btn_dynamic Change coros launched at runtime.
+btn_event Pushbutton event interface.
+\x1b[39m
 '''
 print(tests)
 
@@ -36,6 +40,15 @@ async def pulse(led, ms):
     await asyncio.sleep_ms(ms)
     led.off()
 
+# Pulse an LED when an event triggered
+async def evt_pulse(event, led):
+    while True:
+        event.clear()
+        await event.wait()
+        led.on()
+        await asyncio.sleep_ms(500)
+        led.off()
+
 # Toggle an LED (callback)
 def toggle(led):
     led.toggle()
@@ -94,6 +107,35 @@ def test_swcb():
     sw.open_func(toggle, (green,))
     run(sw)
 
+# Test for the Switch class (events)
+async def do_sw_event():
+    pin = Pin('X1', Pin.IN, Pin.PULL_UP)
+    sw = Switch(pin)
+    sw.open_func(None)
+    sw.close_func(None)
+    tasks = []
+    for event, led in ((sw.close, 1), (sw.open, 2)):
+        tasks.append(asyncio.create_task(evt_pulse(event, LED(led))))
+    await killer(sw)
+    for task in tasks:
+        task.cancel()
+
+def test_sw_event():
+    s = '''
+close pulse red
+open pulses green
+'''
+    print('Test of switch triggering events.')
+    print(helptext)
+    print(s)
+    try:
+        asyncio.run(do_sw_event())
+    except KeyboardInterrupt:
+        print('Interrupted')
+    finally:
+        asyncio.new_event_loop()
+        print(tests)
+
 # Test for the Pushbutton class (coroutines)
 # Pass True to test suppress
 def test_btn(suppress=False, lf=True, df=True):
@@ -181,3 +223,37 @@ def btn_dynamic():
     setup(pb, red, green, yellow, None)
     pb.long_func(setup, (pb, blue, red, green, yellow, 2000))
     run(pb)
+
+# Test for the Pushbutton class (events)
+async def do_btn_event():
+    pin = Pin('X1', Pin.IN, Pin.PULL_UP)
+    pb = Pushbutton(pin)
+    pb.press_func(None)
+    pb.release_func(None)
+    pb.double_func(None)
+    pb.long_func(None)
+    tasks = []
+    for event, led in ((pb.press, 1), (pb.release, 2), (pb.double, 3), (pb.long, 4)):
+        tasks.append(asyncio.create_task(evt_pulse(event, LED(led))))
+    await killer(pb)
+    for task in tasks:
+        task.cancel()
+
+def btn_event():
+    s = '''
+press pulse red
+release pulses green
+double click pulses yellow
+long press pulses blue
+'''
+    print('Test of pushbutton triggering events.')
+    print(helptext)
+    print(s)
+    try:
+        asyncio.run(do_btn_event())
+    except KeyboardInterrupt:
+        print('Interrupted')
+    finally:
+        asyncio.new_event_loop()
+        print(tests)
+