+"""Custom PulseIn Class to read PWM signals"""
+import time
+import subprocess
+import os
+import atexit
+import random
+import struct
+import sysv_ipc
+
+DEBUG = False
+queues = []
+procs = []
+
+# The message queues live outside of python space, and must be formally cleaned!
+def final():
+ """In case the program is cancelled or quit, we need to clean up the PulseIn
+ helper process and also the message queue, this is called at exit to do so"""
+ if DEBUG:
+ print("Cleaning up message queues", queues)
+ print("Cleaning up processes", procs)
+ for q in queues:
+ q.remove()
+ for proc in procs:
+ proc.terminate()
+
+
+atexit.register(final)
+
+# pylint: disable=c-extension-no-member
+class PulseIn:
+ """PulseIn Class to read PWM signals"""
+
+ def __init__(self, pin, maxlen=2, idle_state=False):
+ """Create a PulseIn object associated with the given pin.
+ The object acts as a read-only sequence of pulse lengths with
+ a given max length. When it is active, new pulse lengths are
+ added to the end of the list. When there is no more room
+ (len() == maxlen) the oldest pulse length is removed to make room."""
+ self._pin = pin
+ self._maxlen = maxlen
+ self._idle_state = idle_state
+ self._queue_key = random.randint(1, 9999)
+ try:
+ self._mq = sysv_ipc.MessageQueue(None, flags=sysv_ipc.IPC_CREX)
+ if DEBUG:
+ print("Message Queue Key: ", self._mq.key)
+ queues.append(self._mq)
+ except sysv_ipc.ExistentialError:
+ raise RuntimeError(
+ "Message queue creation failed"
+ ) from sysv_ipc.ExistentialError
+
+ # Check if OS is 64-bit
+ if struct.calcsize("P") * 8 == 64:
+ libgpiod_filename = "libgpiod_pulsein64"
+ else:
+ libgpiod_filename = "libgpiod_pulsein"
+
+ dir_path = os.path.dirname(os.path.realpath(__file__))
+ cmd = [
+ dir_path + "/" + libgpiod_filename,
+ "--pulses",
+ str(maxlen),
+ "--queue",
+ str(self._mq.key),
+ ]
+ if idle_state:
+ cmd.append("-i")
+ cmd.append("gpiochip0")
+ cmd.append(str(pin))
+ if DEBUG:
+ print(cmd)
+
+ self._process = subprocess.Popen(cmd)
+ procs.append(self._process)
+
+ # wait for it to start up
+ if DEBUG:
+ print("Waiting for startup success message from subprocess")
+ message = self._wait_receive_msg(timeout=0.25)
+ if message[0] != b"!":
+ raise RuntimeError("Could not establish message queue with subprocess")
+ self._paused = False
+
+ # pylint: disable=redefined-builtin
+ def _wait_receive_msg(self, timeout=0, type=2):
+ """Internal helper that will wait for new messages of a given type,
+ and throw an exception on timeout"""
+ if timeout > 0:
+ stamp = time.monotonic()
+ while (time.monotonic() - stamp) < timeout:
+ try:
+ message = self._mq.receive(block=False, type=type)
+ return message
+ except sysv_ipc.BusyError:
+ time.sleep(0.001) # wait a bit then retry!
+ # uh-oh timed out
+ raise RuntimeError(
+ "Timed out waiting for PulseIn message. Make sure libgpiod is installed."
+ )
+ message = self._mq.receive(block=True, type=type)
+ return message
+
+ # pylint: enable=redefined-builtin
+
+ def deinit(self):
+ """Deinitialises the PulseIn and releases any hardware and software
+ resources for reuse."""
+ # Clean up after ourselves
+ self._process.terminate()
+ procs.remove(self._process)
+ self._mq.remove()
+ queues.remove(self._mq)
+
+ def __enter__(self):
+ """No-op used by Context Managers."""
+ return self
+
+ def __exit__(self, exc_type, exc_value, tb):
+ """Automatically deinitializes the hardware when exiting a context."""
+ self.deinit()
+
+ def resume(self, trigger_duration=0):
+ """Resumes pulse capture after an optional trigger pulse."""
+ if trigger_duration != 0:
+ self._mq.send("t%d" % trigger_duration, True, type=1)
+ else:
+ self._mq.send("r", True, type=1)
+ self._paused = False
+
+ def pause(self):
+ """Pause pulse capture"""
+ self._mq.send("p", True, type=1)
+ self._paused = True
+
+ @property
+ def paused(self):
+ """True when pulse capture is paused as a result of pause() or
+ an error during capture such as a signal that is too fast."""
+ return self._paused
+
+ @property
+ def maxlen(self):
+ """The maximum length of the PulseIn. When len() is equal to maxlen,
+ it is unclear which pulses are active and which are idle."""
+ return self._maxlen
+
+ def clear(self):
+ """Clears all captured pulses"""
+ self._mq.send("c", True, type=1)
+
+ def popleft(self):
+ """Removes and returns the oldest read pulse."""
+ self._mq.send("^", True, type=1)
+ message = self._wait_receive_msg()
+ reply = int(message[0].decode("utf-8"))
+ # print(reply)
+ if reply == -1:
+ raise IndexError("pop from empty list")
+ return reply
+
+ def __len__(self):
+ """Returns the current pulse length"""
+ self._mq.send("l", True, type=1)
+ message = self._wait_receive_msg()
+ return int(message[0].decode("utf-8"))
+
+ # pylint: disable=redefined-builtin
+ def __getitem__(self, index, type=None):
+ """Returns the value at the given index or values in slice."""
+ self._mq.send("i%d" % index, True, type=1)
+ message = self._wait_receive_msg()
+ ret = int(message[0].decode("utf-8"))
+ if ret == -1:
+ raise IndexError("list index out of range")
+ return ret
+
+ # pylint: enable=redefined-builtin
projects / Adafruit_Blinka-hackapet.git / commitdiff