src/adafruit_blinka/microcontroller/bcm283x/pulseio/PulseIn.py

   1 """Custom PulseIn Class to read PWM signals"""
   2 import time
   3 import subprocess
   4 import os
   5 import atexit
   6 import random
   7 import sysv_ipc
   8
   9 DEBUG = False
  10 queues = []
  11 procs = []
  12
  13 # The message queues live outside of python space, and must be formally cleaned!
  14 def final():
  15     """In case the program is cancelled or quit, we need to clean up the PulseIn
  16     helper process and also the message queue, this is called at exit to do so"""
  17     if DEBUG:
  18         print("Cleaning up message queues", queues)
  19         print("Cleaning up processes", procs)
  20     for q in queues:
  21         q.remove()
  22     for proc in procs:
  23         proc.terminate()
  24
  25
  26 atexit.register(final)
  27
  28 # pylint: disable=c-extension-no-member
  29 class PulseIn:
  30     """PulseIn Class to read PWM signals"""
  31
  32     def __init__(self, pin, maxlen=2, idle_state=False):
  33         """Create a PulseIn object associated with the given pin.
  34         The object acts as a read-only sequence of pulse lengths with
  35         a given max length. When it is active, new pulse lengths are
  36         added to the end of the list. When there is no more room
  37         (len() == maxlen) the oldest pulse length is removed to make room."""
  38         self._pin = pin
  39         self._maxlen = maxlen
  40         self._idle_state = idle_state
  41         self._queue_key = random.randint(1, 9999)
  42         try:
  43             self._mq = sysv_ipc.MessageQueue(None, flags=sysv_ipc.IPC_CREX)
  44             if DEBUG:
  45                 print("Message Queue Key: ", self._mq.key)
  46             queues.append(self._mq)
  47         except sysv_ipc.ExistentialError:
  48             raise RuntimeError(
  49                 "Message queue creation failed"
  50             ) from sysv_ipc.ExistentialError
  51
  52         dir_path = os.path.dirname(os.path.realpath(__file__))
  53         cmd = [
  54             dir_path + "/libgpiod_pulsein",
  55             "--pulses",
  56             str(maxlen),
  57             "--queue",
  58             str(self._mq.key),
  59         ]
  60         if idle_state:
  61             cmd.append("-i")
  62         cmd.append("gpiochip0")
  63         cmd.append(str(pin))
  64         if DEBUG:
  65             print(cmd)
  66
  67         self._process = subprocess.Popen(cmd)
  68         procs.append(self._process)
  69
  70         # wait for it to start up
  71         if DEBUG:
  72             print("Waiting for startup success message from subprocess")
  73         message = self._wait_receive_msg()
  74         if message[0] != b"!":
  75             raise RuntimeError("Could not establish message queue with subprocess")
  76         self._paused = False
  77
  78     # pylint: disable=redefined-builtin
  79     def _wait_receive_msg(self, timeout=0.25, type=2):
  80         """Internal helper that will wait for new messages of a given type,
  81         and throw an exception on timeout"""
  82         stamp = time.monotonic()
  83         while (time.monotonic() - stamp) < timeout:
  84             try:
  85                 message = self._mq.receive(block=False, type=type)
  86                 return message
  87             except sysv_ipc.BusyError:
  88                 time.sleep(0.001)  # wait a bit then retry!
  89         # uh-oh timed out
  90         raise RuntimeError(
  91             "Timed out waiting for PulseIn message. Make sure libgpiod is installed."
  92         )
  93
  94     # pylint: enable=redefined-builtin
  95
  96     def deinit(self):
  97         """Deinitialises the PulseIn and releases any hardware and software
  98         resources for reuse."""
  99         # Clean up after ourselves
 100         self._process.terminate()
 101         procs.remove(self._process)
 102         self._mq.remove()
 103         queues.remove(self._mq)
 104
 105     def __enter__(self):
 106         """No-op used by Context Managers."""
 107         return self
 108
 109     def __exit__(self, exc_type, exc_value, tb):
 110         """Automatically deinitializes the hardware when exiting a context."""
 111         self.deinit()
 112
 113     def resume(self, trigger_duration=0):
 114         """Resumes pulse capture after an optional trigger pulse."""
 115         if trigger_duration != 0:
 116             self._mq.send("t%d" % trigger_duration, True, type=1)
 117         else:
 118             self._mq.send("r", True, type=1)
 119         self._paused = False
 120
 121     def pause(self):
 122         """Pause pulse capture"""
 123         self._mq.send("p", True, type=1)
 124         self._paused = True
 125
 126     @property
 127     def paused(self):
 128         """True when pulse capture is paused as a result of pause() or
 129         an error during capture such as a signal that is too fast."""
 130         return self._paused
 131
 132     @property
 133     def maxlen(self):
 134         """The maximum length of the PulseIn. When len() is equal to maxlen,
 135         it is unclear which pulses are active and which are idle."""
 136         return self._maxlen
 137
 138     def clear(self):
 139         """Clears all captured pulses"""
 140         self._mq.send("c", True, type=1)
 141
 142     def popleft(self):
 143         """Removes and returns the oldest read pulse."""
 144         self._mq.send("^", True, type=1)
 145         message = self._wait_receive_msg()
 146         reply = int(message[0].decode("utf-8"))
 147         # print(reply)
 148         if reply == -1:
 149             raise IndexError("pop from empty list")
 150         return reply
 151
 152     def __len__(self):
 153         """Returns the current pulse length"""
 154         self._mq.send("l", True, type=1)
 155         message = self._wait_receive_msg()
 156         return int(message[0].decode("utf-8"))
 157
 158     # pylint: disable=redefined-builtin
 159     def __getitem__(self, index, type=None):
 160         """Returns the value at the given index or values in slice."""
 161         self._mq.send("i%d" % index, True, type=1)
 162         message = self._wait_receive_msg()
 163         ret = int(message[0].decode("utf-8"))
 164         if ret == -1:
 165             raise IndexError("list index out of range")
 166         return ret
 167
 168     # pylint: enable=redefined-builtin