src/adafruit_blinka/microcontroller/amlogic/a311d/pulseio/PulseIn.py

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