primitives/__init__.py

   1 # __init__.py Common functions for uasyncio primitives
   2
   3 # Copyright (c) 2018-2022 Peter Hinch
   4 # Released under the MIT License (MIT) - see LICENSE file
   5
   6 try:
   7     import uasyncio as asyncio
   8 except ImportError:
   9     import asyncio
  10
  11
  12 async def _g():
  13     pass
  14 type_coro = type(_g())
  15
  16 # If a callback is passed, run it and return.
  17 # If a coro is passed initiate it and return.
  18 # coros are passed by name i.e. not using function call syntax.
  19 def launch(func, tup_args):
  20     res = func(*tup_args)
  21     if isinstance(res, type_coro):
  22         res = asyncio.create_task(res)
  23     return res
  24
  25 def set_global_exception():
  26     def _handle_exception(loop, context):
  27         import sys
  28         sys.print_exception(context["exception"])
  29         sys.exit()
  30     loop = asyncio.get_event_loop()
  31     loop.set_exception_handler(_handle_exception)
  32
  33 _attrs = {
  34     "AADC": "aadc",
  35     "Barrier": "barrier",
  36     "Condition": "condition",
  37     "Delay_ms": "delay_ms",
  38     "Encode": "encoder_async",
  39     "Pushbutton": "pushbutton",
  40     "ESP32Touch": "pushbutton",
  41     "Queue": "queue",
  42     "Semaphore": "semaphore",
  43     "BoundedSemaphore": "semaphore",
  44     "Switch": "switch",
  45     "WaitAll": "events",
  46     "WaitAny": "events",
  47     "ESwitch": "events",
  48     "EButton": "events",
  49     "RingbufQueue": "ringbuf_queue",
  50 }
  51
  52 # Copied from uasyncio.__init__.py
  53 # Lazy loader, effectively does:
  54 #   global attr
  55 #   from .mod import attr
  56 def __getattr__(attr):
  57     mod = _attrs.get(attr, None)
  58     if mod is None:
  59         raise AttributeError(attr)
  60     value = getattr(__import__(mod, None, None, True, 1), attr)
  61     globals()[attr] = value
  62     return value