Untangle code and remove pylint disables

[Adafruit_Blinka-hackapet.git] / src / adafruit_blinka / microcontroller / bcm283x / pulseio / PulseIn.py

diff --git a/src/adafruit_blinka/microcontroller/bcm283x/pulseio/PulseIn.py b/src/adafruit_blinka/microcontroller/bcm283x/pulseio/PulseIn.py
index d2390f6fed489d77f040b9735408ef42981d9084..78d5761afcfb6df5cc0191f28d8e240c7b945eb9 100644 (file)
--- a/src/adafruit_blinka/microcontroller/bcm283x/pulseio/PulseIn.py
+++ b/src/adafruit_blinka/microcontroller/bcm283x/pulseio/PulseIn.py
@@ -1,15 +1,20 @@
+# SPDX-FileCopyrightText: 2021 Melissa LeBlanc-Williams for Adafruit Industries
+#
+# SPDX-License-Identifier: MIT

 """Custom PulseIn Class to read PWM signals"""
 import time
 import subprocess
 import os
 import atexit
 import random
 """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 = []
 
 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
 # 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


@@ -25,6 +30,7 @@ def final():
 
 atexit.register(final)
 
 
 atexit.register(final)
 

+

 # pylint: disable=c-extension-no-member
 class PulseIn:
     """PulseIn Class to read PWM signals"""
 # pylint: disable=c-extension-no-member
 class PulseIn:
     """PulseIn Class to read PWM signals"""


@@ -45,11 +51,19 @@ class PulseIn:
                 print("Message Queue Key: ", self._mq.key)
             queues.append(self._mq)
         except sysv_ipc.ExistentialError:
                 print("Message Queue Key: ", self._mq.key)
             queues.append(self._mq)
         except sysv_ipc.ExistentialError:

-            raise RuntimeError("Message queue creation failed")
+            raise RuntimeError(
+                "Message queue creation failed"
+            ) from sysv_ipc.ExistentialError
+
+        # Check if OS is 64-bit
+        if struct.calcsize("P") * 8 == 64:  # pylint: disable=no-member
+            libgpiod_filename = "libgpiod_pulsein64"
+        else:
+            libgpiod_filename = "libgpiod_pulsein"

 
         dir_path = os.path.dirname(os.path.realpath(__file__))
         cmd = [
 
         dir_path = os.path.dirname(os.path.realpath(__file__))
         cmd = [

-            dir_path + "/libgpiod_pulsein",
+            dir_path + "/" + libgpiod_filename,

             "--pulses",
             str(maxlen),
             "--queue",
             "--pulses",
             str(maxlen),
             "--queue",


@@ -57,37 +71,44 @@ class PulseIn:
         ]
         if idle_state:
             cmd.append("-i")
         ]
         if idle_state:
             cmd.append("-i")

-        cmd.append("gpiochip0")
-        cmd.append(str(pin))
+        if isinstance(pin.id, tuple):
+            cmd.append(f"gpiochip{pin.id[0]}")
+            cmd.append(str(pin.id[1]))
+        else:
+            cmd.append("gpiochip0")
+            cmd.append(str(pin))

         if DEBUG:
             print(cmd)
 
         if DEBUG:
             print(cmd)
 

-        self._process = subprocess.Popen(cmd)
+        self._process = subprocess.Popen(cmd)  # pylint: disable=consider-using-with

         procs.append(self._process)
 
         # wait for it to start up
         if DEBUG:
             print("Waiting for startup success message from subprocess")
         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()
+        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
         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.25, type=2):
+    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"""
         """Internal helper that will wait for new messages of a given type,
         and throw an exception on timeout"""

-        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."
-        )
+        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
 
 
     # pylint: enable=redefined-builtin