primitives/delay_ms.py

   1 # delay_ms.py Now uses ThreadSafeFlag and has extra .wait() API
   2 # Usage:
   3 # from primitives import Delay_ms
   4
   5 # Copyright (c) 2018-2022 Peter Hinch
   6 # Released under the MIT License (MIT) - see LICENSE file
   7
   8 import uasyncio as asyncio
   9 from utime import ticks_add, ticks_diff, ticks_ms
  10 from . import launch
  11
  12 class Delay_ms:
  13
  14     class DummyTimer:  # Stand-in for the timer class. Can be cancelled.
  15         def cancel(self):
  16             pass
  17     _fake = DummyTimer()
  18
  19     def __init__(self, func=None, args=(), duration=1000):
  20         self._func = func
  21         self._args = args
  22         self._durn = duration  # Default duration
  23         self._retn = None  # Return value of launched callable
  24         self._tend = None  # Stop time (absolute ms).
  25         self._busy = False
  26         self._trig = asyncio.ThreadSafeFlag()
  27         self._tout = asyncio.Event()  # Timeout event
  28         self.wait = self._tout.wait  # Allow: await wait_ms.wait()
  29         self.clear = self._tout.clear
  30         self.set = self._tout.set
  31         self._ttask = self._fake  # Timer task
  32         self._mtask = asyncio.create_task(self._run()) #Main task
  33
  34     async def _run(self):
  35         while True:
  36             await self._trig.wait()  # Await a trigger
  37             self._ttask.cancel()  # Cancel and replace
  38             await asyncio.sleep_ms(0)
  39             dt = max(ticks_diff(self._tend, ticks_ms()), 0)  # Beware already elapsed.
  40             self._ttask = asyncio.create_task(self._timer(dt))
  41
  42     async def _timer(self, dt):
  43         await asyncio.sleep_ms(dt)
  44         self._tout.set()  # Only gets here if not cancelled.
  45         self._busy = False
  46         if self._func is not None:
  47             self._retn = launch(self._func, self._args)
  48
  49 # API
  50     # trigger may be called from hard ISR.
  51     def trigger(self, duration=0):  # Update absolute end time, 0-> ctor default
  52         if self._mtask is None:
  53             raise RuntimeError("Delay_ms.deinit() has run.")
  54         self._tend = ticks_add(ticks_ms(), duration if duration > 0 else self._durn)
  55         self._retn = None  # Default in case cancelled.
  56         self._busy = True
  57         self._trig.set()
  58
  59     def stop(self):
  60         self._ttask.cancel()
  61         self._ttask = self._fake
  62         self._busy = False
  63         self._tout.clear()
  64
  65     def __call__(self):  # Current running status
  66         return self._busy
  67
  68     running = __call__
  69
  70     def rvalue(self):
  71         return self._retn
  72
  73     def callback(self, func=None, args=()):
  74         self._func = func
  75         self._args = args
  76
  77     def deinit(self):
  78         self.stop()
  79         self._mtask.cancel()
  80         self._mtask = None