refactor<correct import names for common folder

Author: Ahelsamahy

.vscode/settings.json

@@ -1,6 +1,8 @@
 {
     "cSpell.words": [
+      "BYODR",
       "GPIO",
+      "Jetson",
       "teleop"
     ],
     "python.analysis.typeCheckingMode": "off",

BYODR_utils/JETSON_specific/gpio_relay.py

@@ -2,10 +2,10 @@
 
 import threading
 
-import Jetson.GPIO as GPIO
+import Jetson.GPIO as GPIO  # type: ignore
 
 
-class ThreadSafeGpioRelay(object):
+class ThreadSafeJetsonGpioRelay(object):
     """
     Thread-safe class for managing a GPIO relay on a Jetson Nano.
     """

jetson_runtime/pilot/pilot/tests.py renamed to archived/PIL/tests.py

@@ -8,9 +8,9 @@
 
 from app import CommandProcessor
 from app import PilotApplication
-from byodr.utils import timestamp
-from byodr.utils.navigate import ReloadableDataSource, FileSystemRouteDataSource
-from byodr.utils.testing import CollectPublisher, QueueReceiver, CollectServer
+from BYODR_utils.common import timestamp
+from BYODR_utils.common.navigate import ReloadableDataSource, FileSystemRouteDataSource
+from BYODR_utils.common.testing import CollectPublisher, QueueReceiver, CollectServer
 
 
 def test_create_and_setup(tmpdir):

jetson_runtime/pilot/pilot/tests_relay.py renamed to archived/PIL/tests_relay.py

@@ -8,9 +8,9 @@
 from six.moves import range
 from six.moves.configparser import SafeConfigParser
 
-from byodr.utils import timestamp
-from byodr.utils.testing import QueueReceiver, CollectServer, CollectJSONClient
-from byodr.utils.usbrelay import SearchUsbRelayFactory
+from BYODR_utils.common import timestamp
+from BYODR_utils.common.testing import QueueReceiver, CollectServer, CollectJSONClient
+from BYODR_utils.common.usbrelay import SearchUsbRelayFactory
 from relay import MonitorApplication
 
 

archived/exr/app.py

@@ -18,8 +18,8 @@
 from can import CanError
 from pyueye import ueye
 
-from byodr.utils import timestamp
-from byodr.utils.ipc import ReceiverThread, JSONPublisher, ImagePublisher
+from BYODR_utils.common import timestamp
+from BYODR_utils.common.ipc import ReceiverThread, JSONPublisher, ImagePublisher
 from camera import Camera, FrameThread
 
 logger = logging.getLogger(__name__)

archived/odometry.py

@@ -5,8 +5,8 @@
 
 from gpiozero import DigitalInputDevice
 
-from byodr.utils import timestamp
-from byodr.utils.ipc import JSONPublisher
+from BYODR_utils.common import timestamp
+from BYODR_utils.common.ipc import JSONPublisher
 
 logger = logging.getLogger(__name__)
 log_format = "%(levelname)s: %(filename)s %(funcName)s %(message)s"

archived/tests_rover.py

@@ -3,7 +3,7 @@
 from ConfigParser import SafeConfigParser
 
 from app import RoverApplication
-from byodr.utils.testing import CollectPublisher, QueueReceiver, CollectServer
+from BYODR_utils.common.testing import CollectPublisher, QueueReceiver, CollectServer
 
 
 def test_rover_create_and_setup(tmpdir):

jetson_runtime/BYODR_utils/JETSON_specific/gpio_relay.py

@@ -0,0 +1,41 @@
+from __future__ import absolute_import
+
+import threading
+
+import Jetson.GPIO as GPIO  # type: ignore
+
+
+class ThreadSafeJetsonGpioRelay(object):
+    """
+    Thread-safe class for managing a GPIO relay on a Jetson Nano.
+    """
+
+    def __init__(self, pin=15):
+        self.pin = pin
+        self.state = False  # False for OFF, True for ON
+        self.lock = threading.Lock()
+        GPIO.setmode(GPIO.BOARD)  # Set the pin numbering system to BOARD
+        GPIO.setup(self.pin, GPIO.OUT, initial=GPIO.LOW)
+
+    def open(self):
+        """Turns the relay ON (sets the GPIO pin LOW)."""
+        with self.lock:
+            GPIO.output(self.pin, GPIO.LOW)
+            self.state = False
+
+    def close(self):
+        """Turns the relay OFF (sets the GPIO pin HIGH)."""
+        with self.lock:
+            GPIO.output(self.pin, GPIO.HIGH)
+            self.state = True
+
+    def toggle(self):
+        """Toggles the relay state."""
+        with self.lock:
+            self.state = not self.state
+            GPIO.output(self.pin, GPIO.LOW if self.state else GPIO.HIGH)
+
+    def states(self):
+        """Returns the current state of the relay."""
+        with self.lock:
+            return self.state

jetson_runtime/BYODR_utils/JETSON_specific/utilities.py

@@ -0,0 +1,20 @@
+import subprocess
+
+
+class Nano:
+    @staticmethod
+    def get_ip_address():
+        try:
+            ip_addresses = (
+                subprocess.check_output(
+                    "hostname -I | awk '{for (i=1; i<=NF; i++) if ($i ~ /^192\\.168\\./) print $i}'",
+                    shell=True,
+                )
+                .decode()
+                .strip()
+            )
+            # Split in case there are multiple local IP addresses
+            return ip_addresses
+        except subprocess.CalledProcessError as e:
+            print(f"An error occurred: {e}")
+            return None

jetson_runtime/BYODR_utils/PI_specific/gpio_relay.py

@@ -0,0 +1,48 @@
+from __future__ import absolute_import
+
+import threading
+
+import RPi.GPIO as GPIO  # type: ignore
+
+
+class ThreadSafePi4GpioRelay:
+    """Thread-safe class for managing a GPIO relay on a Raspberry Pi."""
+
+    def __init__(self, pin=15):
+        self.pin = pin
+        self.state = False  # False for OFF, True for ON
+        self.lock = threading.Lock()
+        GPIO.setmode(GPIO.BOARD)  # Set the pin numbering system to BOARD
+        GPIO.setwarnings(False)
+        GPIO.setup(self.pin, GPIO.OUT, initial=GPIO.LOW)
+
+    def open(self):
+        """Turns the relay ON (sets the GPIO pin LOW)."""
+        with self.lock:
+            print("opened the relay")
+            GPIO.output(self.pin, GPIO.LOW)
+            self.state = False
+
+    def close(self):
+        """Turns the relay OFF (sets the GPIO pin HIGH)."""
+        with self.lock:
+            GPIO.output(self.pin, GPIO.HIGH)
+            current_state = GPIO.input(self.pin)
+            print(f"closed the relay, current state: {current_state}")
+            self.state = True
+
+    def toggle(self):
+        """Toggles the relay state."""
+        with self.lock:
+            self.state = not self.state
+            GPIO.output(self.pin, GPIO.LOW if self.state else GPIO.HIGH)
+
+    def get_state(self):
+        """Returns the current state of the relay."""
+        with self.lock:
+            return self.state
+
+    def cleanup(self):
+        """Cleans up the GPIO state."""
+        GPIO.cleanup(self.pin)  # Reset the specific pin before setup
+        GPIO.setup(self.pin, GPIO.OUT, initial=GPIO.LOW)