remove .envrc sourcing of install

Headless too

.gitmodules

@@ -1,3 +1,6 @@
 [submodule "src/astra_description"]
 	path = src/astra_descriptions
 	url = ../astra_descriptions
+[submodule "src/astra_msgs"]
+	path = src/astra_msgs
+	url = git@github.com:SHC-ASTRA/astra_msgs.git

README.md

@@ -16,6 +16,9 @@ You will use these packages to launch all rover-side ROS2 nodes.
   - [Connecting the GuliKit Controller](#connecting-the-gulikit-controller)
 - [Common Problems/Toubleshooting](#common-problemstroubleshooting)
 - [Packages](#packages)
+- [Graphs](#graphs)
+  - [Full System](#full-system)
+  - [Individual Nodes](#individual-nodes)
 - [Maintainers](#maintainers)
 
 ## Software Prerequisites
@@ -140,6 +143,40 @@ A: To find a microcontroller to talk to, Anchor sends a ping to every Serial por
 - [ros2\_interfaces\_pkg](./src/ros2_interfaces_pkg) - Contains custom message types for communication between basestation and the rover over ROS2. (being renamed to `astra_msgs`).
 - [servo\_arm\_twist\_pkg](./src/servo_arm_twist_pkg) - A temporary node to translate controller state from `ros2_joy` to `Twist` messages to control the Arm via IK.
 
+## Graphs
+
+### Full System
+
+> **Anchor stand-alone** (`ros2 launch anchor_pkg rover.launch.py`)
+>
+> ![rqt_graph of Anchor by itself, ran with command: ros2 launch anchor_pkg rover.launch.py](./docs-resources/graph-anchor-standalone.png)
+
+> **Anchor with [basestation-classic](https://github.com/SHC-ASTRA/basestation-classic)**
+>
+> ![rqt_graph of Anchor ran with the same command as above, talking to basestation-classic](./docs-resources/graph-anchor-w-basestation-classic.png)
+
+> **Anchor with Headless** (`ros2 run headless_pkg headless_full`)
+>
+> ![rqt_graph of Anchor ran with Headless](./docs-resources/graph-anchor-w-headless.png)
+
+### Individual Nodes
+
+> **Anchor** (`ros2 run anchor_pkg anchor`)
+>
+> ![rqt_graph of Anchor node running by itself](./docs-resources/graph-anchor-anchor-standalone.png)
+
+> **Core** (`ros2 run core_pkg core --ros-args -p launch_mode:=anchor`)
+>
+> ![rqt_graph of Core node running by itself](./docs-resources/graph-anchor-core-standalone.png)
+
+> **Arm** (`ros2 run arm_pkg arm --ros-args -p launch_mode:=anchor`)
+>
+> ![rqt_graph of Arm node running by itself](./docs-resources/graph-anchor-arm-standalone.png)
+
+> **Bio** (`ros2 run bio_pkg bio --ros-args -p launch_mode:=anchor`)
+>
+> ![rqt_graph of Bio node running by itself](./docs-resources/graph-anchor-bio-standalone.png)
+
 ## Maintainers
 
 | Name | Email | Discord |

auto_start/auto_start_anchor.sh

@@ -14,8 +14,12 @@ echo "[INFO] Network interface is up!"
 # Your actual ROS node start command goes here
 echo "[INFO] Starting ROS node..."
 
-# Source ROS 2 Humble setup script (if we aren't using nix)
+# Source ROS 2 Humble setup script
-command -v ros2 || source /opt/ros/humble/setup.bash
+if command -v nixos-rebuild; then
+    echo "[INFO] running on NixOS"
+else
+    source /opt/ros/humble/setup.bash
+fi
 
 # Source your workspace setup script
 source $SCRIPT_DIR/../install/setup.bash

auto_start/auto_start_headless_full.sh

@@ -15,7 +15,11 @@ echo "[INFO] Network interface is up!"
 echo "[INFO] Starting ROS node..."
 
 # Source ROS 2 Humble setup script
-command -v ros2 || source /opt/ros/humble/setup.bash
+if command -v nixos-rebuild; then
+    echo "[INFO] running on NixOS"
+else
+    source /opt/ros/humble/setup.bash
+fi
 
 # Source your workspace setup script
 source $SCRIPT_DIR/../install/setup.bash

auto_start/start_rosbag.sh

@@ -17,7 +17,11 @@ done
 echo "[INFO] Network interface is up!"
 
 
-command -v ros2 || source /opt/ros/humble/setup.bash
+if command -v nixos-rebuild; then
+    echo "[INFO] running on NixOS"
+else
+    source /opt/ros/humble/setup.bash
+fi
 source $ANCHOR_WS/install/setup.bash
 [[ -f $AUTONOMY_WS/install/setup.bash ]] && source $AUTONOMY_WS/install/setup.bash
 

flake.lock

@@ -1,29 +1,5 @@
 {
   "nodes": {
-    "astra-msgs": {
-      "inputs": {
-        "nix-ros-overlay": "nix-ros-overlay",
-        "nixpkgs": [
-          "astra-msgs",
-          "nix-ros-overlay",
-          "nixpkgs"
-        ]
-      },
-      "locked": {
-        "lastModified": 1771845042,
-        "narHash": "sha256-pGsb93ZlMhP3biy8S2eJc1sW+35vmaxlWHTbuwZDlQI=",
-        "owner": "SHC-ASTRA",
-        "repo": "astra_msgs",
-        "rev": "acabfd117d9711afc420612375b4e02f4ce4982d",
-        "type": "github"
-      },
-      "original": {
-        "owner": "SHC-ASTRA",
-        "repo": "astra_msgs",
-        "rev": "acabfd117d9711afc420612375b4e02f4ce4982d",
-        "type": "github"
-      }
-    },
     "flake-utils": {
       "inputs": {
         "systems": "systems"
@@ -42,55 +18,17 @@
         "type": "github"
       }
     },
-    "flake-utils_2": {
-      "inputs": {
-        "systems": "systems_2"
-      },
-      "locked": {
-        "lastModified": 1731533236,
-        "narHash": "sha256-l0KFg5HjrsfsO/JpG+r7fRrqm12kzFHyUHqHCVpMMbI=",
-        "owner": "numtide",
-        "repo": "flake-utils",
-        "rev": "11707dc2f618dd54ca8739b309ec4fc024de578b",
-        "type": "github"
-      },
-      "original": {
-        "owner": "numtide",
-        "repo": "flake-utils",
-        "type": "github"
-      }
-    },
     "nix-ros-overlay": {
       "inputs": {
         "flake-utils": "flake-utils",
         "nixpkgs": "nixpkgs"
       },
       "locked": {
-        "lastModified": 1770622967,
+        "lastModified": 1770108954,
-        "narHash": "sha256-1LYjTugPSCa/5NkP6/dcZLH5TQYj3R8mAZ/9dgd7jDM=",
+        "narHash": "sha256-VBj6bd4LPPSfsZJPa/UPPA92dOs6tmQo0XZKqfz/3W4=",
         "owner": "lopsided98",
         "repo": "nix-ros-overlay",
-        "rev": "d1b9f17eba909116356436d46b5192d299c6b49a",
+        "rev": "3d05d46451b376e128a1553e78b8870c75d7753a",
-        "type": "github"
-      },
-      "original": {
-        "owner": "lopsided98",
-        "ref": "develop",
-        "repo": "nix-ros-overlay",
-        "type": "github"
-      }
-    },
-    "nix-ros-overlay_2": {
-      "inputs": {
-        "flake-utils": "flake-utils_2",
-        "nixpkgs": "nixpkgs_2"
-      },
-      "locked": {
-        "lastModified": 1771404745,
-        "narHash": "sha256-UVP3TsQJ4PezyQG3B1SsgsTxz32XBVzplJ/cgq7v/uk=",
-        "owner": "lopsided98",
-        "repo": "nix-ros-overlay",
-        "rev": "7cdf7b44ff186869baedbb3b82e5b409bb3e8dd9",
         "type": "github"
       },
       "original": {
@@ -116,30 +54,14 @@
         "type": "github"
       }
     },
-    "nixpkgs_2": {
-      "locked": {
-        "lastModified": 1759381078,
-        "narHash": "sha256-gTrEEp5gEspIcCOx9PD8kMaF1iEmfBcTbO0Jag2QhQs=",
-        "owner": "NixOS",
-        "repo": "nixpkgs",
-        "rev": "7df7ff7d8e00218376575f0acdcc5d66741351ee",
-        "type": "github"
-      },
-      "original": {
-        "owner": "lopsided98",
-        "ref": "nix-ros",
-        "repo": "nixpkgs",
-        "type": "github"
-      }
-    },
     "root": {
       "inputs": {
-        "astra-msgs": "astra-msgs",
+        "nix-ros-overlay": "nix-ros-overlay",
-        "nix-ros-overlay": "nix-ros-overlay_2",
         "nixpkgs": [
           "nix-ros-overlay",
           "nixpkgs"
-        ]
+        ],
+        "treefmt-nix": "treefmt-nix"
       }
     },
     "systems": {
@@ -157,18 +79,23 @@
         "type": "github"
       }
     },
-    "systems_2": {
+    "treefmt-nix": {
+      "inputs": {
+        "nixpkgs": [
+          "nixpkgs"
+        ]
+      },
       "locked": {
-        "lastModified": 1681028828,
+        "lastModified": 1773297127,
-        "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
+        "narHash": "sha256-6E/yhXP7Oy/NbXtf1ktzmU8SdVqJQ09HC/48ebEGBpk=",
-        "owner": "nix-systems",
+        "owner": "numtide",
-        "repo": "default",
+        "repo": "treefmt-nix",
-        "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
+        "rev": "71b125cd05fbfd78cab3e070b73544abe24c5016",
         "type": "github"
       },
       "original": {
-        "owner": "nix-systems",
+        "owner": "numtide",
-        "repo": "default",
+        "repo": "treefmt-nix",
         "type": "github"
       }
     }

flake.nix

@@ -3,13 +3,21 @@
 
   inputs = {
     nix-ros-overlay.url = "github:lopsided98/nix-ros-overlay/develop";
-    nixpkgs.follows = "nix-ros-overlay/nixpkgs";
+    nixpkgs.follows = "nix-ros-overlay/nixpkgs"; # IMPORTANT!!!
-    astra-msgs.url =
+
-      "github:SHC-ASTRA/astra_msgs/acabfd117d9711afc420612375b4e02f4ce4982d";
+    treefmt-nix = {
+      url = "github:numtide/treefmt-nix";
+      inputs.nixpkgs.follows = "nixpkgs";
+    };
   };
 
   outputs =
-    { self, nix-ros-overlay, nixpkgs, astra-msgs }:
+    {
+      self,
+      nix-ros-overlay,
+      nixpkgs,
+      ...
+    }@inputs:
     nix-ros-overlay.inputs.flake-utils.lib.eachDefaultSystem (
       system:
       let
@@ -17,31 +25,27 @@
           inherit system;
           overlays = [ nix-ros-overlay.overlays.default ];
         };
-
-        astra_msgs_pkgs = astra-msgs.packages.${system};
-        rosDistro = "humble";
       in
       {
         devShells.default = pkgs.mkShell {
           name = "ASTRA Anchor";
-
           packages = with pkgs; [
             colcon
-            astra_msgs_pkgs.astra-msgs
+            (python313.withPackages (
-
+              p: with p; [
-            (pkgs.rosPackages.${rosDistro}.python3.withPackages (p: with p; [
                 pyserial
+                python-can
                 pygame
                 scipy
                 crccheck
                 black
-            ]))
+              ]
-
+            ))
-            (with rosPackages.${rosDistro};
+            (
+              with rosPackages.humble;
               buildEnv {
                 paths = [
                   ros-core
-                  rqt-graph
                   ros2cli
                   ros2run
                   ros2bag
@@ -77,24 +81,22 @@
                   ros2-controllers
                   chomp-motion-planner
                 ];
-              })
+              }
+            )
           ];
-
-          env = {
-            ASTRAMSGS = "${astra-msgs.outPath}";
-          };
-
           shellHook = ''
+            # Display stuff
             export DISPLAY=''${DISPLAY:-:0}
             export QT_X11_NO_MITSHM=1
           '';
         };
+
+        formatter = (inputs.treefmt-nix.lib.evalModule pkgs ./treefmt.nix).config.build.wrapper;
       }
     );
 
   nixConfig = {
     extra-substituters = [ "https://ros.cachix.org" ];
-    extra-trusted-public-keys =
+    extra-trusted-public-keys = [ "ros.cachix.org-1:dSyZxI8geDCJrwgvCOHDoAfOm5sV1wCPjBkKL+38Rvo=" ];
-      [ "ros.cachix.org-1:dSyZxI8geDCJrwgvCOHDoAfOm5sV1wCPjBkKL+38Rvo=" ];
   };
 }

src/anchor_pkg/anchor_pkg/anchor_node.py

@@ -1,29 +1,26 @@
 import rclpy
 from rclpy.node import Node
 from rclpy.executors import ExternalShutdownException
-from std_srvs.srv import Empty
+from rcl_interfaces.msg import ParameterDescriptor, ParameterType
 
 import signal
-import time
 import atexit
 
-import serial
+from .connector import (
-import serial.tools.list_ports
+    Connector,
-import os
+    MockConnector,
+    SerialConnector,
+    CANConnector,
+    NoValidDeviceException,
+    NoWorkingDeviceException,
+)
+from .convert import string_to_viccan
 import sys
 import threading
-import glob
 
 from std_msgs.msg import String, Header
 from astra_msgs.msg import VicCAN
 
-KNOWN_USBS = [
-    (0x2E8A, 0x00C0),  # Raspberry Pi Pico
-    (0x1A86, 0x55D4),  # Adafruit Feather ESP32 V2
-    (0x10C4, 0xEA60),  # DOIT ESP32 Devkit V1
-    (0x1A86, 0x55D3),  # ESP32 S3 Development Board
-]
-
 
 class Anchor(Node):
     """
@@ -42,132 +39,61 @@ class Anchor(Node):
         - For testing without an actual MCU, publish strings here as if they came from an MCU
     * /anchor/to_vic/relay
         - Core, Arm, and Bio publish VicCAN messages to this topic to send to the MCU
-    * /anchor/to_vic/relay_string
-        - Publish raw strings to this topic to send directly to the MCU for debugging
     """
 
+    connector: Connector
+
     def __init__(self):
-        # Initalize node with name
+        super().__init__("anchor_node")
-        super().__init__("anchor_node")  # previously 'serial_publisher'
 
-        self.serial_port: str | None = None  # e.g., "/dev/ttyUSB0"
+        logger = self.get_logger()
 
-        # Serial port override
+        self.declare_parameter(
-        if port_override := os.getenv("PORT_OVERRIDE"):
+            "connector",
-            self.serial_port = port_override
+            "auto",
-
+            ParameterDescriptor(
-        ##################################################
+                name="connector",
-        # Serial MCU Discovery
+                description="Declares which MCU connector should be used. Defaults to 'auto'.",
-
+                type=ParameterType.PARAMETER_STRING,
-        # If there was not a port override, look for a MCU over USB for Serial.
+                additional_constraints="Must be 'serial', 'can', 'mock', or 'auto'.",
-        if self.serial_port is None:
+            ),
-            comports = serial.tools.list_ports.comports()
-            real_ports = list(
-                filter(
-                    lambda p: p.vid is not None
-                    and p.pid is not None
-                    and p.device is not None,
-                    comports,
         )
-            )
-            recog_ports = list(filter(lambda p: (p.vid, p.pid) in KNOWN_USBS, comports))
 
-            if len(recog_ports) == 1:  # Found singular recognized MCU
+        # Determine which connector to use. Options are Mock, Serial, and CAN
-                found_port = recog_ports[0]
+        connector_select = (
-                self.get_logger().info(
+            self.get_parameter("connector").get_parameter_value().string_value
-                    f"Selecting MCU '{found_port.description}' at {found_port.device}."
         )
-                self.serial_port = found_port.device  # String, location of device file; e.g., '/dev/ttyACM0'
-            elif len(recog_ports) > 1:  # Found multiple recognized MCUs
-                # Kinda jank log message
-                self.get_logger().error(
-                    f"Found multiple recognized MCUs: {[p.device for p in recog_ports].__str__()}"
-                )
-                # Don't set self.serial_port; later if-statement will exit()
-            elif (
-                len(recog_ports) == 0 and len(real_ports) > 0
-            ):  # Found real ports but none recognized; i.e. maybe found an IMU or camera but not a MCU
-                self.get_logger().error(
-                    f"No recognized MCUs found; instead found {[p.device for p in real_ports].__str__()}."
-                )
-                # Don't set self.serial_port; later if-statement will exit()
-            else:  # Found jack shit
-                self.get_logger().error("No valid Serial ports specified or found.")
-                # Don't set self.serial_port; later if-statement will exit()
 
-        # We still don't have a serial port; fall back to legacy discovery (Areeb's code)
+        match connector_select:
-        # Loop through all serial devices on the computer to check for the MCU
+            case "serial":
-        if self.serial_port is None:
+                logger.info("using serial connector")
-            self.get_logger().warning("Falling back to legacy MCU discovery...")
+                self.connector = SerialConnector(self.get_logger())
-            ports = Anchor.list_serial_ports()
+            case "can":
-            for _ in range(4):
+                logger.info("using CAN connector")
-                if self.serial_port is not None:
+                self.connector = CANConnector(self.get_logger())
-                    break
+            case "mock":
-                for port in ports:
+                logger.info("using mock connector")
+                self.connector = MockConnector(self.get_logger())
+            case "auto":
+                logger.info("automatically determining connector")
                 try:
-                        # connect and send a ping command
+                    logger.info("trying CAN connector")
-                        ser = serial.Serial(port, 115200, timeout=1)
+                    self.connector = CANConnector(self.get_logger())
-                        # (f"Checking port {port}...")
+                except (NoValidDeviceException, NoWorkingDeviceException, TypeError):
-                        ser.write(b"ping\n")
+                    logger.info("CAN connector failed, trying serial connector")
-                        response = ser.read_until(bytes("\n", "utf8"))
+                    self.connector = SerialConnector(self.get_logger())
-
+            case _:
-                        # if pong is in response, then we are talking with the MCU
+                self.get_logger().fatal(
-                        if b"pong" in response:
+                    f"invalid value for connector parameter: {connector_select}"
-                            self.serial_port = port
-                            self.get_logger().info(f"Found MCU at {self.serial_port}!")
-                            break
-                    except:
-                        pass
-
-        # If port is still None then we ain't finding no mcu
-        if self.serial_port is None:
-            self.get_logger().error("Unable to find MCU. Exiting...")
-            time.sleep(1)
-            sys.exit(1)
-        # Found a Serial port, try to open it; above code has not officially opened a Serial port
-        else:
-            self.get_logger().debug(
-                f"Attempting to open Serial port '{self.serial_port}'..."
-            )
-            try:
-                self.serial_interface = serial.Serial(
-                    self.serial_port, 115200, timeout=1
                 )
+                exit(1)
 
-                # Attempt to get name of connected MCU
+        # Close devices on exit
-                self.serial_interface.write(
-                    b"can_relay_mode,on\n"
-                )  # can_relay_ready,[mcu]
-                mcu_name: str = ""
-                for _ in range(4):
-                    response = self.serial_interface.read_until(bytes("\n", "utf8"))
-                    try:
-                        if b"can_relay_ready" in response:
-                            args: list[str] = response.decode("utf8").strip().split(",")
-                            if len(args) == 2:
-                                mcu_name = args[1]
-                            break
-                    except UnicodeDecodeError:
-                        pass  # ignore malformed responses
-                self.get_logger().info(
-                    f"MCU '{mcu_name}' is ready at '{self.serial_port}'."
-                )
-
-            except serial.SerialException as e:
-                self.get_logger().error(
-                    f"Could not open Serial port '{self.serial_port}' for reason:"
-                )
-                self.get_logger().error(e.strerror)
-                time.sleep(1)
-                sys.exit(1)
-
-        # Close serial port on exit
         atexit.register(self.cleanup)
 
-        ##################################################
         # ROS2 Topic Setup
 
-        # New pub/sub with VicCAN
+        # Publishers
         self.fromvic_debug_pub_ = self.create_publisher(
             String, "/anchor/from_vic/debug", 20
         )
@@ -181,163 +107,45 @@ class Anchor(Node):
             VicCAN, "/anchor/from_vic/bio", 20
         )
 
+        # Subscribers
+        self.tovic_sub_ = self.create_subscription(
+            VicCAN, "/anchor/to_vic/relay", self.connector.write, 20
+        )
         self.mock_mcu_sub_ = self.create_subscription(
             String, "/anchor/from_vic/mock_mcu", self.on_mock_fromvic, 20
         )
-        self.tovic_sub_ = self.create_subscription(
-            VicCAN, "/anchor/to_vic/relay", self.on_relay_tovic_viccan, 20
-        )
-        self.tovic_debug_sub_ = self.create_subscription(
-            String, "/anchor/to_vic/relay_string", self.on_relay_tovic_string, 20
-        )
 
-        # Create publishers
+    def cleanup(self):
-        self.arm_pub = self.create_publisher(String, "/anchor/arm/feedback", 10)
+        self.connector.cleanup()
-        self.core_pub = self.create_publisher(String, "/anchor/core/feedback", 10)
-        self.bio_pub = self.create_publisher(String, "/anchor/bio/feedback", 10)
-
-        self.debug_pub = self.create_publisher(String, "/anchor/debug", 10)
-
-        # Create a subscriber
-        self.relay_sub = self.create_subscription(
-            String, "/anchor/relay", self.on_relay_tovic_string, 10
-        )
 
     def read_MCU(self):
         """Check the USB serial port for new data from the MCU, and publish string to appropriate topics"""
-        try:
+        output = self.connector.read()
-            output = str(self.serial_interface.readline(), "utf8")
 
-            if output:
+        if not output:
-                self.relay_fromvic(output)
-                # All output over debug temporarily
-                # self.get_logger().info(f"[MCU] {output}")
-                msg = String()
-                msg.data = output
-                self.debug_pub.publish(msg)
-                if output.startswith("can_relay_fromvic,core"):
-                    self.core_pub.publish(msg)
-                elif output.startswith("can_relay_fromvic,arm") or output.startswith(
-                    "can_relay_fromvic,digit"
-                ):  # digit for voltage readings
-                    self.arm_pub.publish(msg)
-                if output.startswith("can_relay_fromvic,citadel") or output.startswith(
-                    "can_relay_fromvic,digit"
-                ):  # digit for SHT sensor
-                    self.bio_pub.publish(msg)
-                # msg = String()
-                # msg.data = output
-                # self.debug_pub.publish(msg)
             return
-        except serial.SerialException as e:
+
-            print(f"SerialException: {e}")
+        self.relay_fromvic(output)
-            print("Closing serial port.")
+
-            try:
+    def relay_fromvic(self, msg: VicCAN):
-                if self.serial_interface.is_open:
+        """Relay a string message from the MCU to the appropriate VicCAN topic"""
-                    self.serial_interface.close()
+        msg.header = Header(stamp=self.get_clock().now().to_msg(), frame_id="from_vic")
-            except:
+
-                pass
+        if msg.mcu_name == "core":
-            exit(1)
+            self.fromvic_core_pub_.publish(msg)
-        except TypeError as e:
+        elif msg.mcu_name == "arm" or msg.mcu_name == "digit":
-            print(f"TypeError: {e}")
+            self.fromvic_arm_pub_.publish(msg)
-            print("Closing serial port.")
+        elif msg.mcu_name == "citadel" or msg.mcu_name == "digit":
-            try:
+            self.fromvic_bio_pub_.publish(msg)
-                if self.serial_interface.is_open:
-                    self.serial_interface.close()
-            except:
-                pass
-            exit(1)
-        except Exception as e:
-            print(f"Exception: {e}")
-            # print("Closing serial port.")
-            # if self.ser.is_open:
-            #     self.ser.close()
-            # exit(1)
 
     def on_mock_fromvic(self, msg: String):
-        """For testing without an actual MCU, publish strings here as if they came from an MCU"""
+        viccan = string_to_viccan(
-        # self.get_logger().info(f"Got command from mock MCU: {msg}")
+            msg.data,
-        self.relay_fromvic(msg.data)
+            "mock",
-
+            self.get_logger(),
-    def on_relay_tovic_viccan(self, msg: VicCAN):
-        """Relay a VicCAN message to the MCU"""
-        output: str = f"can_relay_tovic,{msg.mcu_name},{msg.command_id}"
-        for num in msg.data:
-            output += f",{round(num, 7)}"  # limit to 7 decimal places
-        output += "\n"
-        # self.get_logger().info(f"VicCAN relay to MCU: {output}")
-        self.serial_interface.write(bytes(output, "utf8"))
-
-    def relay_fromvic(self, msg: str):
-        """Relay a string message from the MCU to the appropriate VicCAN topic"""
-        self.fromvic_debug_pub_.publish(String(data=msg))
-        parts = msg.strip().split(",")
-        if len(parts) > 0 and parts[0] != "can_relay_fromvic":
-            self.get_logger().debug(f"Ignoring non-VicCAN message: '{msg.strip()}'")
-            return
-
-        # String validation
-        malformed: bool = False
-        malformed_reason: str = ""
-        if len(parts) < 3 or len(parts) > 7:
-            malformed = True
-            malformed_reason = (
-                f"invalid argument count (expected [3,7], got {len(parts)})"
         )
-        elif parts[1] not in ["core", "arm", "digit", "citadel", "broadcast"]:
+        if viccan:
-            malformed = True
+            self.relay_fromvic(viccan)
-            malformed_reason = f"invalid mcu_name '{parts[1]}'"
-        elif not (parts[2].isnumeric()) or int(parts[2]) < 0:
-            malformed = True
-            malformed_reason = f"command_id '{parts[2]}' is not a non-negative integer"
-        else:
-            for x in parts[3:]:
-                try:
-                    float(x)
-                except ValueError:
-                    malformed = True
-                    malformed_reason = f"data '{x}' is not a float"
-                    break
-
-        if malformed:
-            self.get_logger().warning(
-                f"Ignoring malformed from_vic message: '{msg.strip()}'; reason: {malformed_reason}"
-            )
-            return
-
-        # Have valid VicCAN message
-
-        output = VicCAN()
-        output.mcu_name = parts[1]
-        output.command_id = int(parts[2])
-        if len(parts) > 3:
-            output.data = [float(x) for x in parts[3:]]
-        output.header = Header(
-            stamp=self.get_clock().now().to_msg(), frame_id="from_vic"
-        )
-
-        # self.get_logger().info(f"Relaying from MCU: {output}")
-        if output.mcu_name == "core":
-            self.fromvic_core_pub_.publish(output)
-        elif output.mcu_name == "arm" or output.mcu_name == "digit":
-            self.fromvic_arm_pub_.publish(output)
-        elif output.mcu_name == "citadel" or output.mcu_name == "digit":
-            self.fromvic_bio_pub_.publish(output)
-
-    def on_relay_tovic_string(self, msg: String):
-        """Relay a raw string message to the MCU for debugging"""
-        message = msg.data
-        # self.get_logger().info(f"Sending command to MCU: {msg}")
-        self.serial_interface.write(bytes(message, "utf8"))
-
-    @staticmethod
-    def list_serial_ports():
-        return glob.glob("/dev/ttyUSB*") + glob.glob("/dev/ttyACM*")
-
-    def cleanup(self):
-        print("Cleaning up before terminating...")
-        if self.serial_interface.is_open:
-            self.serial_interface.close()
 
 
 def main(args=None):

src/anchor_pkg/anchor_pkg/connector.py

@@ -0,0 +1,423 @@
+from abc import ABC, abstractmethod
+from astra_msgs.msg import VicCAN
+from rclpy.impl.rcutils_logger import RcutilsLogger
+from .convert import string_to_viccan
+
+# CAN
+import can
+import can.interfaces.socketcan
+import struct
+
+# Serial
+import serial
+import serial.tools.list_ports
+
+KNOWN_USBS = [
+    (0x2E8A, 0x00C0),  # Raspberry Pi Pico
+    (0x1A86, 0x55D4),  # Adafruit Feather ESP32 V2
+    (0x10C4, 0xEA60),  # DOIT ESP32 Devkit V1
+    (0x1A86, 0x55D3),  # ESP32 S3 Development Board
+]
+BAUD_RATE = 115200
+
+
+class NoValidDeviceException(Exception):
+    pass
+
+
+class NoWorkingDeviceException(Exception):
+    pass
+
+
+class MultipleValidDevicesException(Exception):
+    pass
+
+
+class DeviceClosedException(Exception):
+    pass
+
+
+class Connector(ABC):
+    logger: RcutilsLogger
+
+    def __init__(self, logger: RcutilsLogger):
+        self.logger = logger
+
+    @abstractmethod
+    def read(self) -> VicCAN | None:
+        pass
+
+    @abstractmethod
+    def write(self, msg: VicCAN):
+        pass
+
+    def cleanup(self):
+        pass
+
+
+class SerialConnector(Connector):
+    port: str
+    mcu_name: str
+    serial_interface: serial.Serial
+    override: bool
+
+    def __init__(self, logger: RcutilsLogger):
+        super().__init__(logger)
+
+        ports = self._find_ports()
+
+        if len(ports) <= 0:
+            raise NoValidDeviceException("no valid serial device found")
+        if (l := len(ports)) > 1:
+            raise MultipleValidDevicesException(
+                f"too many ({l}) valid serial devices found"
+            )
+
+        # check each of our ports to make sure one of them is responding
+        port = ports[0]
+        mcu_name = self._get_name(port)
+        if not mcu_name:
+            raise NoWorkingDeviceException(
+                f"found {port}, but it did not respond with its name"
+            )
+
+        self.port = port
+        self.mcu_name = mcu_name
+
+        # if we fail at this point, it should crash because we've already tested the port
+        self.serial_interface = serial.Serial(self.port, BAUD_RATE, timeout=1)
+
+    def _find_ports(self) -> list[str]:
+        """
+        Finds all valid ports but does not test them
+
+        returns: all valid ports
+        """
+        comports = serial.tools.list_ports.comports()
+        valid_ports = list(
+            map(  # get just device strings
+                lambda p: p.device,
+                filter(  # make sure we have a known device
+                    lambda p: (p.vid, p.pid) in KNOWN_USBS and p.device is not None,
+                    comports,
+                ),
+            )
+        )
+        self.logger.info(f"found valid MCU ports: [ {', '.join(valid_ports)} ]")
+        return valid_ports
+
+    def _get_name(self, port: str) -> str | None:
+        """
+        Get the name of the MCU (if it works)
+
+        returns: str name of the MCU, None if it doesn't work
+        """
+        # attempt to open the serial port
+        serial_interface: serial.Serial
+        try:
+            self.logger.info(f"asking {port} for its name")
+            serial_interface = serial.Serial(port, BAUD_RATE, timeout=1)
+
+            serial_interface.write(b"can_relay_mode,on\n")
+
+            for i in range(4):
+                self.logger.debug(f"attempt {i + 1} of 4 asking {port} for its name")
+                response = serial_interface.read_until(bytes("\n", "utf8"))
+                try:
+                    if b"can_relay_ready" in response:
+                        args: list[str] = response.decode("utf8").strip().split(",")
+                        if len(args) == 2:
+                            self.logger.info(f"we are talking to {args[1]}")
+                            return args[1]
+                        break
+                except UnicodeDecodeError as e:
+                    self.logger.info(
+                        f"ignoring UnicodeDecodeError when asking for MCU name: {e}"
+                    )
+
+            if serial_interface.is_open:
+                serial_interface.close()
+        except serial.SerialException as e:
+            self.logger.error(f"SerialException when asking for MCU name: {e}")
+
+        return None
+
+    def read(self) -> VicCAN | None:
+        try:
+            raw = str(self.serial_interface.readline(), "utf8")
+
+            if not raw:
+                return None
+
+            return string_to_viccan(raw, self.mcu_name, self.logger)
+        except serial.SerialException as e:
+            self.logger.error(f"SerialException: {e}")
+            raise DeviceClosedException(f"serial port {self.port} closed unexpectedly")
+        except TypeError as e:
+            self.logger.error(f"TypeError: {e}")
+            raise DeviceClosedException(f"serial port {self.port} closed unexpectedly")
+        except Exception:
+            pass  # pretty much no other error matters
+
+    def write(self, msg: VicCAN):
+        # go from [ w, x, y, z ] -> "w,x,y,z" & round to 7 digits max
+        data = ",".join([str(round(x, 7)) for x in msg.data])
+        output = f"can_relay_tovic,{msg.mcu_name},{msg.command_id},{data}\n"
+        self.serial_interface.write(bytes(output, "utf8"))
+
+    def cleanup(self):
+        self.logger.info(f"closing serial port if open {self.port}")
+        try:
+            if self.serial_interface.is_open:
+                self.serial_interface.close()
+        except Exception as e:
+            self.logger.error(e)
+
+
+class CANConnector(Connector):
+    def __init__(self, logger: RcutilsLogger):
+        super().__init__(logger)
+
+        self.can_channel: str | None = None
+        self.can_bus: can.BusABC | None = None
+
+        if self.can_channel is None:
+            avail = can.interfaces.socketcan.SocketcanBus._detect_available_configs()
+
+            if len(avail) == 0:
+                raise NoValidDeviceException("no CAN interfaces found")
+            if (l := len(avail)) > 1:
+                channels = ", ".join(str(b.get("channel")) for b in avail)
+                raise MultipleValidDevicesException(
+                    f"too many ({l}) CAN interfaces found: [{channels}]"
+                )
+
+            bus = avail[0]
+            self.can_channel = str(bus.get("channel"))
+            self.logger.info(f"found CAN interface '{self.can_channel}'")
+
+        try:
+            self.can_bus = can.Bus(
+                interface="socketcan",
+                channel=self.can_channel,
+                bitrate=1_000_000,
+            )
+        except can.CanError as e:
+            raise NoWorkingDeviceException(
+                f"could not open CAN channel '{self.can_channel}': {e}"
+            )
+
+        if self.can_channel and self.can_channel.startswith("v"):
+            self.logger.warn("using virtual CAN interface; this is likely vcan*")
+
+    def read(self) -> VicCAN | None:
+        if not self.can_bus:
+            raise DeviceClosedException("CAN bus not initialized")
+
+        try:
+            message = self.can_bus.recv(timeout=0.0)
+        except can.CanError as e:
+            self.logger.error(f"CAN error while receiving: {e}")
+            raise DeviceClosedException("CAN bus closed unexpectedly")
+
+        if message is None:
+            return None
+
+        arbitration_id = message.arbitration_id & 0x7FF
+        data_bytes = bytes(message.data)
+
+        mcu_key = (arbitration_id >> 8) & 0b111
+        data_type_key = (arbitration_id >> 6) & 0b11
+        command = arbitration_id & 0x3F
+
+        key_to_mcu: dict[int, str] = {
+            1: "broadcast",
+            2: "core",
+            3: "arm",
+            4: "digit",
+            5: "faerie",
+            6: "citadel",
+        }
+
+        mcu_name = key_to_mcu.get(mcu_key)
+        if mcu_name is None:
+            self.logger.warn(
+                f"received CAN frame with unknown MCU key {mcu_key}; id=0x{arbitration_id:X}"
+            )
+            return None
+
+        data: list[float] = []
+
+        try:
+            if data_type_key == 3:
+                data = []
+            elif data_type_key == 0:
+                if len(data_bytes) < 8:
+                    self.logger.warn(
+                        f"received double payload with insufficient length {len(data_bytes)}; dropping frame"
+                    )
+                    return None
+                (value,) = struct.unpack("<d", data_bytes[:8])
+                data = [float(value)]
+            elif data_type_key == 1:
+                if len(data_bytes) < 8:
+                    self.logger.warn(
+                        f"received float32x2 payload with insufficient length {len(data_bytes)}; dropping frame"
+                    )
+                    return None
+                v1, v2 = struct.unpack("<ff", data_bytes[:8])
+                data = [float(v1), float(v2)]
+            elif data_type_key == 2:
+                if len(data_bytes) < 8:
+                    self.logger.warn(
+                        f"received int16x4 payload with insufficient length {len(data_bytes)}; dropping frame"
+                    )
+                    return None
+                i1, i2, i3, i4 = struct.unpack("<hhhh", data_bytes[:8])
+                data = [float(i1), float(i2), float(i3), float(i4)]
+            else:
+                self.logger.warn(
+                    f"received CAN frame with unknown data_type_key {data_type_key}; id=0x{arbitration_id:X}"
+                )
+                return None
+        except struct.error as e:
+            self.logger.error(f"error unpacking CAN payload: {e}")
+            return None
+
+        viccan = VicCAN(
+            mcu_name=mcu_name,
+            command_id=int(command),
+            data=data,
+        )
+
+        self.logger.debug(
+            f"received CAN frame id=0x{message.arbitration_id:X}, "
+            f"decoded as VicCAN(mcu_name={viccan.mcu_name}, command_id={viccan.command_id}, data={viccan.data})"
+        )
+
+        return viccan
+
+    def write(self, msg: VicCAN):
+        if not self.can_bus:
+            raise DeviceClosedException("CAN bus not initialized")
+
+        # Build 11-bit arbitration ID according to the VicCAN scheme:
+        # bits 10..8: targeted MCU key
+        # bits 7..6:  data type key
+        # bits 5..0:  command
+
+        # Map MCU name to 3-bit key.
+        mcu_name = (msg.mcu_name or "").lower()
+        mcu_key_map: dict[str, int] = {
+            "broadcast": 1,
+            "core": 2,
+            "arm": 3,
+            "digit": 4,
+            "faerie": 5,
+            "citadel": 6,
+        }
+
+        if mcu_name not in mcu_key_map:
+            self.logger.error(
+                f"unknown VicCAN mcu_name '{msg.mcu_name}' for CAN frame; dropping message"
+            )
+            return
+
+        mcu_key = mcu_key_map[mcu_name] & 0b111
+
+        # Infer data type key from payload length according to the table:
+        # 0: double x1, 1: float32 x2, 2: int16 x4, 3: empty
+        data_len = len(msg.data)
+        if data_len == 0:
+            data_type_key = 3
+        elif data_len == 1:
+            data_type_key = 0
+        elif data_len == 2:
+            data_type_key = 1
+        elif data_len == 4:
+            data_type_key = 2
+        else:
+            # Fallback: treat any other non-zero length as float32 x2
+            self.logger.warn(
+                f"unexpected VicCAN data length {data_len}; encoding as float32 x2 (key=1) and truncating/padding as needed"
+            )
+            data_type_key = 1
+
+        # Command is limited to 6 bits.
+        command = int(msg.command_id)
+        if command < 0:
+            self.logger.error(f"invalid negative command_id for CAN frame: {command}")
+            return
+        if command > 0x3F:
+            self.logger.warn(
+                f"command_id 0x{command:X} exceeds 6-bit range; truncating to lower 6 bits"
+            )
+            command &= 0x3F
+
+        arbitration_id = (
+            ((mcu_key & 0b111) << 8) | ((data_type_key & 0b11) << 6) | (command & 0x3F)
+        )
+
+        # Map VicCAN.data (floats) to up to 8 CAN data bytes.
+        raw_bytes: list[int] = []
+        for value in msg.data:
+            try:
+                b = int(round(value))
+            except (TypeError, ValueError):
+                self.logger.error(
+                    f"non-numeric VicCAN data value: {value}; dropping message"
+                )
+                return
+
+            if b < 0 or b > 255:
+                self.logger.warn(
+                    f"VicCAN data value {value} out of byte range; clamping into [0, 255]"
+                )
+                b = max(0, min(255, b))
+
+            raw_bytes.append(b)
+
+        if len(raw_bytes) > 8:
+            self.logger.warn(
+                f"VicCAN data too long for single CAN frame ({len(raw_bytes)} > 8); truncating"
+            )
+            raw_bytes = raw_bytes[:8]
+
+        try:
+            can_message = can.Message(
+                arbitration_id=arbitration_id,
+                data=raw_bytes,
+                is_extended_id=False,
+            )
+        except Exception as e:
+            self.logger.error(f"failed to construct CAN message: {e}")
+            return
+
+        try:
+            self.can_bus.send(can_message)
+            self.logger.debug(
+                f"sent CAN frame id=0x{can_message.arbitration_id:X}, "
+                f"data={list(can_message.data)}"
+            )
+        except can.CanError as e:
+            self.logger.error(f"CAN error while sending: {e}")
+            raise DeviceClosedException("CAN bus closed unexpectedly")
+
+    def cleanup(self):
+        try:
+            if self.can_bus is not None:
+                self.logger.info("shutting down CAN bus")
+                self.can_bus.shutdown()
+        except Exception as e:
+            self.logger.error(e)
+
+
+class MockConnector(Connector):
+    def __init__(self, logger: RcutilsLogger):
+        super().__init__(logger)
+
+    def read(self) -> VicCAN | None:
+        return None
+
+    def write(self, msg: VicCAN):
+        print(msg)

src/anchor_pkg/anchor_pkg/convert.py

@@ -0,0 +1,41 @@
+from astra_msgs.msg import VicCAN
+from rclpy.impl.rcutils_logger import RcutilsLogger
+
+
+def string_to_viccan(msg: str, mcu_name: str, logger: RcutilsLogger):
+    """
+    Converts the serial string VicCAN format to a ROS2 VicCAN message.
+    Does not fill out the Header of the message.
+    On a failure, it will log at a debug level why it failed.
+
+    Parameters:
+      * msg: str
+        - The message in serial VicCAN format
+      * mcu_name: str
+        - The name of the MCU (e.g. core, citadel, arm)
+      * logger: RcutilsLogger
+        - A logger retrieved from node.get_logger()
+
+    Returns:
+      * VicCAN | None
+        - The VicCAN message on a success or None on a failure
+    """
+
+    parts: list[str] = msg.split(",")
+
+    # don't need an extra check because len of .split output is always >= 1
+    if parts[0] != "can_relay_fromvic":
+        logger.debug(f"got non-CAN data from {mcu_name}: {msg}")
+        return None
+    elif len(parts) < 3:
+        logger.debug(f"got garbage (not enough parts) CAN data from {mcu_name}: {msg}")
+        return None
+    elif len(parts) > 7:
+        logger.debug(f"got garbage (too many parts) CAN data from {mcu_name}: {msg}")
+        return None
+
+    return VicCAN(
+        mcu_name=parts[1],
+        command_id=int(parts[2]),
+        data=[float(x) for x in parts[3:]],
+    )

src/anchor_pkg/launch/rover.launch.py

@@ -1,136 +1,91 @@
-#!/usr/bin/env python3
-
 from launch import LaunchDescription
-from launch.actions import DeclareLaunchArgument, OpaqueFunction, Shutdown
+from launch.actions import DeclareLaunchArgument, Shutdown
-from launch.substitutions import (
+from launch.conditions import IfCondition
-    LaunchConfiguration,
+from launch.substitutions import LaunchConfiguration
-    ThisLaunchFileDir,
-    PathJoinSubstitution,
-)
 from launch_ros.actions import Node
 
 
-# Prevent making __pycache__ directories
+def generate_launch_description():
-from sys import dont_write_bytecode
+    connector = LaunchConfiguration("connector")
+    use_ptz = LaunchConfiguration("use_ptz")
 
-dont_write_bytecode = True
+    ld = LaunchDescription()
 
+    # arguments
+    ld.add_action(
+        DeclareLaunchArgument(
+            "connector",
+            default_value="auto",
+            description="Connector parameter for anchor node (default: auto)",
+        )
+    )
 
-def launch_setup(context, *args, **kwargs):
+    ld.add_action(
-    # Retrieve the resolved value of the launch argument 'mode'
+        DeclareLaunchArgument(
-    mode = LaunchConfiguration("mode").perform(context)
+            "use_ptz",
-    nodes = []
+            default_value="true",  # must be string for launch system
+            description="Whether to launch PTZ node (default: true)",
+        )
+    )
 
-    if mode == "anchor":
+    # nodes
-        # Launch every node and pass "anchor" as the parameter
+    ld.add_action(
-
-        nodes.append(
-            Node(
-                package="arm_pkg",
-                executable="arm",  # change as needed
-                name="arm",
-                output="both",
-                parameters=[{"launch_mode": mode}],
-                on_exit=Shutdown(),
-            )
-        )
-        nodes.append(
-            Node(
-                package="core_pkg",
-                executable="core",  # change as needed
-                name="core",
-                output="both",
-                parameters=[{"launch_mode": mode}],
-                on_exit=Shutdown(),
-            )
-        )
-        nodes.append(
-            Node(
-                package="core_pkg",
-                executable="ptz",  # change as needed
-                name="ptz",
-                output="both",
-                # Currently don't shutdown all nodes if the PTZ node fails, as it is not critical
-                # on_exit=Shutdown()  # Uncomment if you want to shutdown on PTZ failure
-            )
-        )
-        nodes.append(
-            Node(
-                package="bio_pkg",
-                executable="bio",  # change as needed
-                name="bio",
-                output="both",
-                parameters=[{"launch_mode": mode}],
-                on_exit=Shutdown(),
-            )
-        )
-        nodes.append(
-            Node(
-                package="anchor_pkg",
-                executable="anchor",  # change as needed
-                name="anchor",
-                output="both",
-                parameters=[{"launch_mode": mode}],
-                on_exit=Shutdown(),
-            )
-        )
-    elif mode in ["arm", "core", "bio", "ptz"]:
-        # Only launch the node corresponding to the provided mode.
-        if mode == "arm":
-            nodes.append(
         Node(
             package="arm_pkg",
             executable="arm",
             name="arm",
             output="both",
-                    parameters=[{"launch_mode": mode}],
+            parameters=[{"launch_mode": "anchor"}],
             on_exit=Shutdown(),
         )
     )
-        elif mode == "core":
+
-            nodes.append(
+    ld.add_action(
         Node(
             package="core_pkg",
             executable="core",
             name="core",
             output="both",
-                    parameters=[{"launch_mode": mode}],
+            parameters=[{"launch_mode": "anchor"}],
             on_exit=Shutdown(),
         )
     )
-        elif mode == "bio":
+
-            nodes.append(
+    ld.add_action(
-                Node(
-                    package="bio_pkg",
-                    executable="bio",
-                    name="bio",
-                    output="both",
-                    parameters=[{"launch_mode": mode}],
-                    on_exit=Shutdown(),
-                )
-            )
-        elif mode == "ptz":
-            nodes.append(
         Node(
             package="core_pkg",
             executable="ptz",
             name="ptz",
             output="both",
-                    on_exit=Shutdown(),  # on fail, shutdown if this was the only node to be launched
+            condition=IfCondition(use_ptz),
         )
     )
-    else:
-        # If an invalid mode is provided, print an error.
-        print("Invalid mode provided. Choose one of: arm, core, bio, anchor, ptz.")
 
-    return nodes
+    ld.add_action(
-
+        Node(
-
+            package="bio_pkg",
-def generate_launch_description():
+            executable="bio",
-    declare_arg = DeclareLaunchArgument(
+            name="bio",
-        "mode",
+            output="both",
-        default_value="anchor",
+            parameters=[{"launch_mode": "anchor"}],
-        description="Launch mode: arm, core, bio, anchor, or ptz",
+            on_exit=Shutdown(),
+        )
     )
 
-    return LaunchDescription([declare_arg, OpaqueFunction(function=launch_setup)])
+    ld.add_action(
+        Node(
+            package="anchor_pkg",
+            executable="anchor",
+            name="anchor",
+            output="both",
+            parameters=[
+                {
+                    "launch_mode": "anchor",
+                    "connector": connector,
+                }
+            ],
+            on_exit=Shutdown(),
+        )
+    )
+
+    return ld

src/anchor_pkg/package.xml

@@ -3,13 +3,14 @@
 <package format="3">
   <name>anchor_pkg</name>
   <version>0.0.0</version>
-  <description>TODO: Package description</description>
+  <description>Anchor -- ROS and CAN relay node</description>
-  <maintainer email="tristanmcginnis26@gmail.com">tristan</maintainer>
+  <maintainer email="rjm0037@uah.edu">Riley</maintainer>
   <license>AGPL-3.0-only</license>
 
   <depend>rclpy</depend>
   <depend>common_interfaces</depend>
   <depend>python3-serial</depend>
+  <depend>python3-can</depend>
 
   <build_depend>black</build_depend>
 

src/arm_pkg/arm_pkg/arm_headless.py

@@ -1,279 +0,0 @@
-import rclpy
-from rclpy.node import Node
-
-import pygame
-
-import time
-
-import serial
-import sys
-import threading
-import glob
-import os
-
-from std_msgs.msg import String
-from astra_msgs.msg import ControllerState
-from astra_msgs.msg import ArmManual
-from astra_msgs.msg import ArmIK
-
-
-os.environ["SDL_AUDIODRIVER"] = (
-    "dummy"  # Force pygame to use a dummy audio driver before pygame.init()
-)
-os.environ["SDL_VIDEODRIVER"] = "dummy"  # Prevents pygame from trying to open a display
-
-
-class Headless(Node):
-    def __init__(self):
-        # Initalize node with name
-        super().__init__("arm_headless")
-
-        # Depricated, kept temporarily for reference
-        # self.create_timer(0.20, self.send_controls)#read and send controls
-
-        self.create_timer(0.1, self.send_manual)
-
-        # Create a publisher to publish any output the pico sends
-
-        # Depricated, kept temporarily for reference
-        # self.publisher = self.create_publisher(ControllerState, '/astra/arm/control', 10)
-        self.manual_pub = self.create_publisher(ArmManual, "/arm/control/manual", 10)
-
-        # Create a subscriber to listen to any commands sent for the pico
-
-        # Depricated, kept temporarily for reference
-        # self.subscriber = self.create_subscription(String, '/astra/arm/feedback', self.read_feedback, 10)
-
-        # self.debug_sub = self.create_subscription(String, '/arm/feedback/debug', self.read_feedback, 10)
-
-        self.laser_status = 0
-
-        # Initialize pygame
-        pygame.init()
-
-        # Initialize the gamepad module
-        pygame.joystick.init()
-
-        # Check if any gamepad is connected
-        if pygame.joystick.get_count() == 0:
-            print("No gamepad found.")
-            pygame.quit()
-            exit()
-        for event in pygame.event.get():
-            if event.type == pygame.QUIT:
-                pygame.quit()
-                exit()
-
-        # Initialize the first gamepad, print name to terminal
-        self.gamepad = pygame.joystick.Joystick(0)
-        self.gamepad.init()
-        print(f"Gamepad Found: {self.gamepad.get_name()}")
-        #
-        #
-
-    def run(self):
-        # This thread makes all the update processes run in the background
-        thread = threading.Thread(target=rclpy.spin, args={self}, daemon=True)
-        thread.start()
-
-        try:
-            while rclpy.ok():
-                # Check the pico for updates
-
-                # self.read_feedback()
-                if (
-                    pygame.joystick.get_count() == 0
-                ):  # if controller disconnected, wait for it to be reconnected
-                    print(f"Gamepad disconnected: {self.gamepad.get_name()}")
-
-                    while pygame.joystick.get_count() == 0:
-                        # self.send_controls() #depricated, kept for reference temporarily
-                        self.send_manual()
-                        # self.read_feedback()
-                    self.gamepad = pygame.joystick.Joystick(0)
-                    self.gamepad.init()  # re-initialized gamepad
-                    print(f"Gamepad reconnected: {self.gamepad.get_name()}")
-
-        except KeyboardInterrupt:
-            sys.exit(0)
-
-    def send_manual(self):
-        for event in pygame.event.get():
-            if event.type == pygame.QUIT:
-                pygame.quit()
-                exit()
-        input = ArmManual()
-
-        # Triggers for gripper control
-        if self.gamepad.get_axis(2) > 0:  # left trigger
-            input.gripper = -1
-        elif self.gamepad.get_axis(5) > 0:  # right trigger
-            input.gripper = 1
-
-        # Toggle Laser
-        if self.gamepad.get_button(7):  # Start
-            self.laser_status = 1
-        elif self.gamepad.get_button(6):  # Back
-            self.laser_status = 0
-        input.laser = self.laser_status
-
-        if self.gamepad.get_button(5):  # right bumper, control effector
-
-            # Left stick X-axis for effector yaw
-            if self.gamepad.get_axis(0) > 0:
-                input.effector_yaw = 1
-            elif self.gamepad.get_axis(0) < 0:
-                input.effector_yaw = -1
-
-            # Right stick X-axis for effector roll
-            if self.gamepad.get_axis(3) > 0:
-                input.effector_roll = 1
-            elif self.gamepad.get_axis(3) < 0:
-                input.effector_roll = -1
-
-        else:  # Control arm axis
-            dpad_input = self.gamepad.get_hat(0)
-            input.axis0 = 0
-            if dpad_input[0] == 1:
-                input.axis0 = 1
-            elif dpad_input[0] == -1:
-                input.axis0 = -1
-
-            if self.gamepad.get_axis(0) > 0.15 or self.gamepad.get_axis(0) < -0.15:
-                input.axis1 = round(self.gamepad.get_axis(0))
-
-            if self.gamepad.get_axis(1) > 0.15 or self.gamepad.get_axis(1) < -0.15:
-                input.axis2 = -1 * round(self.gamepad.get_axis(1))
-
-            if self.gamepad.get_axis(4) > 0.15 or self.gamepad.get_axis(4) < -0.15:
-                input.axis3 = -1 * round(self.gamepad.get_axis(4))
-
-            # input.axis1 = -1 * round(self.gamepad.get_axis(0))#left x-axis
-            # input.axis2 = -1 * round(self.gamepad.get_axis(1))#left y-axis
-            # input.axis3 = -1 * round(self.gamepad.get_axis(4))#right y-axis
-
-        # Button Mappings
-        # axis2 -> LT
-        # axis5 -> RT
-        # Buttons0 -> A
-        # Buttons1 -> B
-        # Buttons2 -> X
-        # Buttons3 -> Y
-        # Buttons4 -> LB
-        # Buttons5 -> RB
-        # Buttons6 -> Back
-        # Buttons7 -> Start
-
-        input.linear_actuator = 0
-
-        if pygame.joystick.get_count() != 0:
-
-            self.get_logger().info(
-                f"[Ctrl] {input.axis0}, {input.axis1}, {input.axis2}, {input.axis3}\n"
-            )
-            self.manual_pub.publish(input)
-        else:
-            pass
-
-        pass
-
-    # Depricated, kept temporarily for reference
-    # def send_controls(self):
-
-    #     for event in pygame.event.get():
-    #         if event.type == pygame.QUIT:
-    #             pygame.quit()
-    #             exit()
-    #     input = ControllerState()
-
-    #     input.lt = self.gamepad.get_axis(2)#left trigger
-    #     input.rt = self.gamepad.get_axis(5)#right trigger
-
-    #     #input.lb = self.gamepad.get_button(9)#Value must be converted to bool
-    #     if(self.gamepad.get_button(4)):#left bumper
-    #         input.lb = True
-    #     else:
-    #         input.lb = False
-
-    #     #input.rb = self.gamepad.get_button(10)#Value must be converted to bool
-    #     if(self.gamepad.get_button(5)):#right bumper
-    #         input.rb = True
-    #     else:
-    #         input.rb = False
-
-    #     #input.plus = self.gamepad.get_button(6)#plus button
-    #     if(self.gamepad.get_button(7)):#plus button
-    #         input.plus = True
-    #     else:
-    #         input.plus = False
-
-    #     #input.minus = self.gamepad.get_button(4)#minus button
-    #     if(self.gamepad.get_button(6)):#minus button
-    #         input.minus = True
-    #     else:
-    #         input.minus = False
-
-    #     input.ls_x = round(self.gamepad.get_axis(0),2)#left x-axis
-    #     input.ls_y = round(self.gamepad.get_axis(1),2)#left y-axis
-    #     input.rs_x = round(self.gamepad.get_axis(3),2)#right x-axis
-    #     input.rs_y = round(self.gamepad.get_axis(4),2)#right y-axis
-
-    #     #input.a = self.gamepad.get_button(1)#A button
-    #     if(self.gamepad.get_button(0)):#A button
-    #         input.a = True
-    #     else:
-    #         input.a = False
-    #     #input.b = self.gamepad.get_button(0)#B button
-    #     if(self.gamepad.get_button(1)):#B button
-    #         input.b = True
-    #     else:
-    #         input.b = False
-    #     #input.x = self.gamepad.get_button(3)#X button
-    #     if(self.gamepad.get_button(2)):#X button
-    #         input.x = True
-    #     else:
-    #         input.x = False
-    #     #input.y = self.gamepad.get_button(2)#Y button
-    #     if(self.gamepad.get_button(3)):#Y button
-    #         input.y = True
-    #     else:
-    #         input.y = False
-
-    #     dpad_input = self.gamepad.get_hat(0)#D-pad input
-
-    #     #not using up/down on DPad
-    #     input.d_up = False
-    #     input.d_down = False
-
-    #     if(dpad_input[0] == 1):#D-pad right
-    #         input.d_right = True
-    #     else:
-    #         input.d_right = False
-    #     if(dpad_input[0] == -1):#D-pad left
-    #         input.d_left = True
-    #     else:
-    #         input.d_left = False
-
-    #     if pygame.joystick.get_count() != 0:
-
-    #         self.get_logger().info(f"[Ctrl] Updated Controller State\n")
-
-    #         self.publisher.publish(input)
-    #     else:
-    #         pass
-
-
-def main(args=None):
-    rclpy.init(args=args)
-
-    node = Headless()
-
-    rclpy.spin(node)
-    rclpy.shutdown()
-
-    # tb_bs = BaseStation()
-    # node.run()
-
-
-if __name__ == "__main__":
-    main()

src/arm_pkg/arm_pkg/arm_node.py

@@ -1,181 +1,221 @@
-import rclpy
-from rclpy.node import Node
-import serial
 import sys
 import threading
-import glob
-import time
-import atexit
 import signal
-from std_msgs.msg import String
-from astra_msgs.msg import ArmManual
-from astra_msgs.msg import SocketFeedback
-from astra_msgs.msg import DigitFeedback
-from sensor_msgs.msg import JointState
 import math
+from warnings import deprecated
 
-# control_qos = qos.QoSProfile(
+import rclpy
-#     history=qos.QoSHistoryPolicy.KEEP_LAST,
+from rclpy.node import Node
-#     depth=1,
+from rclpy.executors import ExternalShutdownException
-#     reliability=qos.QoSReliabilityPolicy.BEST_EFFORT,
+from rclpy import qos
-#     durability=qos.QoSDurabilityPolicy.VOLATILE,
+
-#     deadline=1000,
+from std_msgs.msg import String, Header
-#     lifespan=500,
+from sensor_msgs.msg import JointState
+from control_msgs.msg import JointJog
+from astra_msgs.msg import SocketFeedback, DigitFeedback, ArmManual  # TODO: Old topics
+from astra_msgs.msg import ArmFeedback, VicCAN, RevMotorState
+
+control_qos = qos.QoSProfile(
+    history=qos.QoSHistoryPolicy.KEEP_LAST,
+    depth=2,
+    reliability=qos.QoSReliabilityPolicy.BEST_EFFORT,  # Best Effort subscribers are still compatible with Reliable publishers
+    durability=qos.QoSDurabilityPolicy.VOLATILE,
+    # deadline=Duration(seconds=1),
+    # lifespan=Duration(nanoseconds=500_000_000),  # 500ms
     # liveliness=qos.QoSLivelinessPolicy.SYSTEM_DEFAULT,
-#     liveliness_lease_duration=5000
+    # liveliness_lease_duration=Duration(seconds=5),
-# )
+)
 
-serial_pub = None
 thread = None
 
 
-class SerialRelay(Node):
+class ArmNode(Node):
+    """Relay between Anchor and Basestation/Headless/Moveit2 for Arm related topics."""
+
+    # Every non-fixed joint defined in Arm's URDF
+    # Used for JointState and JointJog messsages
+    all_joint_names = [
+        "axis_0_joint",
+        "axis_1_joint",
+        "axis_2_joint",
+        "axis_3_joint",
+        "wrist_yaw_joint",
+        "wrist_roll_joint",
+        "ef_gripper_left_joint",
+    ]
+
+    # Used to verify the length of an incoming VicCAN feedback message
+    # Key is VicCAN command_id, value is expected length of data list
+    viccan_socket_msg_len_dict = {
+        53: 4,
+        54: 4,
+        55: 4,
+        58: 4,
+        59: 4,
+    }
+
+    viccan_digit_msg_len_dict = {
+        54: 4,
+        55: 2,
+        59: 2,
+    }
+
     def __init__(self):
-        # Initialize node
         super().__init__("arm_node")
 
-        # Get launch mode parameter
+        self.get_logger().info(f"arm launch_mode is: anchor")  # Hey I like the output
-        self.declare_parameter("launch_mode", "arm")
-        self.launch_mode = self.get_parameter("launch_mode").value
-        self.get_logger().info(f"arm launch_mode is: {self.launch_mode}")
 
-        # Create publishers
+        ##################################################
-        self.debug_pub = self.create_publisher(String, "/arm/feedback/debug", 10)
+        # Parameters
-        self.socket_pub = self.create_publisher(
-            SocketFeedback, "/arm/feedback/socket", 10
-        )
-        self.digit_pub = self.create_publisher(DigitFeedback, "/arm/feedback/digit", 10)
-        self.feedback_timer = self.create_timer(0.25, self.publish_feedback)
 
-        # Create subscribers
+        self.declare_parameter("use_old_topics", True)
-        self.man_sub = self.create_subscription(
+        self.use_old_topics = (
-            ArmManual, "/arm/control/manual", self.send_manual, 2
+            self.get_parameter("use_old_topics").get_parameter_value().bool_value
         )
 
-        # New messages
+        ##################################################
-        self.joint_state_pub = self.create_publisher(JointState, "joint_states", 10)
+        # Old topics
-        self.joint_state = JointState()
-        self.joint_state.name = [
-            "Axis_0_Joint",
-            "Axis_1_Joint",
-            "Axis_2_Joint",
-            "Axis_3_Joint",
-            "Wrist_Differential_Joint",
-            "Wrist-EF_Roll_Joint",
-            "Gripper_Slider_Left",
-        ]
-        self.joint_state.position = [0.0] * len(
-            self.joint_state.name
-        )  # Initialize with zeros
 
-        self.joint_command_sub = self.create_subscription(
+        if self.use_old_topics:
-            JointState, "/joint_commands", self.joint_command_callback, 10
+            # Anchor topics
-        )
-
-        # Topics used in anchor mode
-        if self.launch_mode == "anchor":
             self.anchor_sub = self.create_subscription(
                 String, "/anchor/arm/feedback", self.anchor_feedback, 10
             )
             self.anchor_pub = self.create_publisher(String, "/anchor/relay", 10)
 
-        self.arm_feedback = SocketFeedback()
+            # Create publishers
-        self.digit_feedback = DigitFeedback()
+            self.socket_pub = self.create_publisher(
-
+                SocketFeedback, "/arm/feedback/socket", 10
-        # Search for ports IF in 'arm' (standalone) and not 'anchor' mode
-        if self.launch_mode == "arm":
-            # Loop through all serial devices on the computer to check for the MCU
-            self.port = None
-            ports = SerialRelay.list_serial_ports()
-            for _ in range(4):
-                for port in ports:
-                    try:
-                        # connect and send a ping command
-                        ser = serial.Serial(port, 115200, timeout=1)
-                        # print(f"Checking port {port}...")
-                        ser.write(b"ping\n")
-                        response = ser.read_until("\n")  # type: ignore
-
-                        # if pong is in response, then we are talking with the MCU
-                        if b"pong" in response:
-                            self.port = port
-                            self.get_logger().info(f"Found MCU at {self.port}!")
-                            break
-                    except:
-                        pass
-                if self.port is not None:
-                    break
-
-            if self.port is None:
-                self.get_logger().info(
-                    "Unable to find MCU... please make sure it is connected."
             )
-                time.sleep(1)
+            self.arm_feedback = SocketFeedback()
-                sys.exit(1)
+            self.digit_pub = self.create_publisher(
+                DigitFeedback, "/arm/feedback/digit", 10
+            )
+            self.digit_feedback = DigitFeedback()
+            self.feedback_timer = self.create_timer(0.25, self.publish_feedback)
 
-            self.ser = serial.Serial(self.port, 115200)
+            # Create subscribers
-            atexit.register(self.cleanup)
+            self.man_sub = self.create_subscription(
+                ArmManual, "/arm/control/manual", self.send_manual, 10
+            )
 
-    def run(self):
+        ###################################################
-        global thread
+        # New topics
-        thread = threading.Thread(target=rclpy.spin, args=(self,), daemon=True)
-        thread.start()
 
-        # if in arm mode, will need to read from the MCU
+        # Anchor topics
 
-        try:
+        # from_vic
-            while rclpy.ok():
+        self.anchor_fromvic_sub_ = self.create_subscription(
-                if self.launch_mode == "arm":
+            VicCAN, "/anchor/from_vic/arm", self.relay_fromvic, 20
-                    if self.ser.in_waiting:
+        )
-                        self.read_mcu()
+        # to_vic
-                    else:
+        self.anchor_tovic_pub_ = self.create_publisher(
-                        time.sleep(0.1)
+            VicCAN, "/anchor/to_vic/relay", 20
-        except KeyboardInterrupt:
+        )
-            pass
-        finally:
-            self.cleanup()
 
-    # Currently will just spit out all values over the /arm/feedback/debug topic as strings
+        # Control
-    def read_mcu(self):
+
-        try:
+        # Manual: /arm/manual/joint_jog is published by Basestation or Headless
-            output = str(self.ser.readline(), "utf8")
+        self.man_jointjog_sub_ = self.create_subscription(
-            if output:
+            JointJog,
-                # self.get_logger().info(f"[MCU] {output}")
+            "/arm/manual/joint_jog",
-                msg = String()
+            self.jointjog_callback,
-                msg.data = output
+            qos_profile=control_qos,
-                self.debug_pub.publish(msg)
+        )
-        except serial.SerialException:
+        # IK: /joint_commands is published by JointTrajectoryController via topic_based_control
-            self.get_logger().info("SerialException caught... closing serial port.")
+        self.joint_command_sub_ = self.create_subscription(
-            if self.ser.is_open:
+            JointState,
-                self.ser.close()
+            "/joint_commands",
-            pass
+            self.joint_command_callback,
-        except TypeError as e:
+            qos_profile=control_qos,
-            self.get_logger().info(f"TypeError: {e}")
+        )
-            print("Closing serial port.")
+
-            if self.ser.is_open:
+        # Feedback
-                self.ser.close()
+
-            pass
+        # Combined Socket and Digit feedback
-        except Exception as e:
+        self.arm_feedback_pub_ = self.create_publisher(
-            print(f"Exception: {e}")
+            ArmFeedback,
-            print("Closing serial port.")
+            "/arm/feedback",
-            if self.ser.is_open:
+            qos_profile=qos.qos_profile_sensor_data,
-                self.ser.close()
+        )
-            pass
+        # IK arm pose: /joint_states is published from here to topic_based_control
+        self.joint_state_pub_ = self.create_publisher(
+            JointState, "/joint_states", qos_profile=qos.qos_profile_sensor_data
+        )
+
+        ###################################################
+        # Saved state
+
+        # Combined Socket and Digit feedback
+        self.arm_feedback_new = ArmFeedback()
+
+        # IK Arm pose
+        self.saved_joint_state = JointState()
+        self.saved_joint_state.name = self.all_joint_names
+        # ... initialize with zeros
+        self.saved_joint_state.position = [0.0] * len(self.saved_joint_state.name)
+        self.saved_joint_state.velocity = [0.0] * len(self.saved_joint_state.name)
+
+    def jointjog_callback(self, msg: JointJog):
+        if len(msg.joint_names) != len(msg.velocities):
+            self.get_logger().debug("Ignoring malformed /arm/manual/joint_jog message.")
+            return
+
+        # Grab velocities from message
+        velocities = [
+            (
+                msg.velocities[msg.joint_names.index(joint_name)]  # type: ignore
+                if joint_name in msg.joint_names
+                else 0.0
+            )
+            for joint_name in self.all_joint_names
+        ]
+        # Deadzone
+        velocities = [vel if abs(vel) > 0.05 else 0.0 for vel in velocities]
+
+        self.send_velocities(velocities, msg.header)
+
+        # TODO: use msg.duration
 
     def joint_command_callback(self, msg: JointState):
-        # Embedded takes deg*10, ROS2 uses Radians
+        if len(msg.position) < 7 and len(msg.velocity) < 7:
-        positions = [math.degrees(pos) * 10 for pos in msg.position]
+            self.get_logger().debug("Ignoring malformed /joint_command message.")
-        # Axis 2 & 3 URDF direction is inverted
+            return  # command needs either position or velocity for all 7 joints
-        positions[2] = -positions[2]
-        positions[3] = -positions[3]
 
-        # Set target angles for each arm axis for embedded IK PID to handle
+        # Grab velocities from message
-        command = f"can_relay_tovic,arm,32,{positions[0]},{positions[1]},{positions[2]},{positions[3]}\n"
+        velocities = [
-        # Wrist yaw and roll
+            (
-        command += f"can_relay_tovic,digit,32,{positions[4]},{positions[5]}\n"
+                msg.velocity[msg.name.index(joint_name)]  # type: ignore
-        # Gripper IK does not have adequate hardware yet
+                if joint_name in msg.name
-        self.send_cmd(command)
+                else 0.0
+            )
+            for joint_name in self.all_joint_names
+        ]
 
+        self.send_velocities(velocities, msg.header)
+
+    def send_velocities(self, velocities: list[float], header: Header):
+        # ROS2's rad/s to VicCAN's deg/s*10; don't convert gripper's m/s
+        velocities = [
+            math.degrees(vel) * 10 if i < 6 else vel for i, vel in enumerate(velocities)
+        ]
+
+        # Send Axis 0-3
+        self.anchor_tovic_pub_.publish(
+            VicCAN(mcu_name="arm", command_id=43, data=velocities[0:3], header=header)
+        )
+        # Send Wrist yaw and roll
+        # TODO: Verify embedded
+        self.anchor_tovic_pub_.publish(
+            VicCAN(mcu_name="digit", command_id=43, data=velocities[4:5], header=header)
+        )
+        # Send End Effector Gripper
+        # TODO: Verify m/s received correctly by embedded
+        self.anchor_tovic_pub_.publish(
+            VicCAN(mcu_name="digit", command_id=26, data=[velocities[6]], header=header)
+        )
+
+    @deprecated("Uses an old message type. Will be removed at some point.")
     def send_manual(self, msg: ArmManual):
         axis0 = msg.axis0
         axis1 = -1 * msg.axis1
@@ -200,24 +240,17 @@ class SerialRelay(Node):
 
         return
 
+    @deprecated("Uses an old message type. Will be removed at some point.")
     def send_cmd(self, msg: str):
-        if (
+        output = String(data=msg)
-            self.launch_mode == "anchor"
-        ):  # if in anchor mode, send to anchor node to relay
-            output = String()
-            output.data = msg
         self.anchor_pub.publish(output)
-        elif self.launch_mode == "arm":  # if in standalone mode, send to MCU directly
-            self.get_logger().info(f"[Arm to MCU] {msg}")
-            self.ser.write(bytes(msg, "utf8"))
 
+    @deprecated("Uses an old message type. Will be removed at some point.")
     def anchor_feedback(self, msg: String):
         output = msg.data
         if output.startswith("can_relay_fromvic,arm,55"):
-            # pass
             self.updateAngleFeedback(output)
         elif output.startswith("can_relay_fromvic,arm,54"):
-            # pass
             self.updateBusVoltage(output)
         elif output.startswith("can_relay_fromvic,arm,53"):
             self.updateMotorFeedback(output)
@@ -235,19 +268,133 @@ class SerialRelay(Node):
             if len(parts) >= 4:
                 self.digit_feedback.wrist_angle = float(parts[3])
                 # self.digit_feedback.wrist_roll = float(parts[4])
-                self.joint_state.position[4] = math.radians(
-                    float(parts[4])
-                )  # Wrist roll
-                self.joint_state.position[5] = math.radians(
-                    float(parts[3])
-                )  # Wrist yaw
         else:
             return
 
+    def relay_fromvic(self, msg: VicCAN):
+        # Code for socket and digit are broken out for cleaner code
+        if msg.mcu_name == "arm":
+            self.process_fromvic_arm(msg)
+        elif msg.mcu_name == "digit":
+            self.process_fromvic_digit(msg)
+
+    def process_fromvic_arm(self, msg: VicCAN):
+        assert msg.mcu_name == "arm"
+
+        # Check message len to prevent crashing on bad data
+        if msg.command_id in self.viccan_socket_msg_len_dict:
+            expected_len = self.viccan_socket_msg_len_dict[msg.command_id]
+            if len(msg.data) != expected_len:
+                self.get_logger().warning(
+                    f"Ignoring VicCAN message with id {msg.command_id} due to unexpected data length (expected {expected_len}, got {len(msg.data)})"
+                )
+                return
+
+        match msg.command_id:
+            case 53:  # REV SPARK MAX feedback
+                motorId = round(msg.data[0])
+                motor: RevMotorState | None = None
+                match motorId:
+                    case 1:
+                        motor = self.arm_feedback_new.axis1_motor
+                    case 2:
+                        motor = self.arm_feedback_new.axis2_motor
+                    case 3:
+                        motor = self.arm_feedback_new.axis3_motor
+                    case 4:
+                        motor = self.arm_feedback_new.axis0_motor
+
+                if motor:
+                    motor.temperature = float(msg.data[1]) / 10.0
+                    motor.voltage = float(msg.data[2]) / 10.0
+                    motor.current = float(msg.data[3]) / 10.0
+                    motor.header.stamp = msg.header.stamp
+
+                self.arm_feedback_pub_.publish(self.arm_feedback_new)
+            case 54:  # Board voltages
+                self.arm_feedback_new.socket_voltage.vbatt = float(msg.data[0]) / 100.0
+                self.arm_feedback_new.socket_voltage.v12 = float(msg.data[1]) / 100.0
+                self.arm_feedback_new.socket_voltage.v5 = float(msg.data[2]) / 100.0
+                self.arm_feedback_new.socket_voltage.v3 = float(msg.data[3]) / 100.0
+                self.arm_feedback_new.socket_voltage.header.stamp = msg.header.stamp
+            case 55:  # Arm joint positions
+                angles = [angle / 10.0 for angle in msg.data]  # VicCAN sends deg*10
+                # Joint state publisher for URDF visualization
+                self.saved_joint_state.position[0] = math.radians(angles[0])  # Axis 0
+                self.saved_joint_state.position[1] = math.radians(angles[1])  # Axis 1
+                self.saved_joint_state.position[2] = math.radians(angles[2])  # Axis 2
+                self.saved_joint_state.position[3] = math.radians(angles[3])  # Axis 3
+                # Wrist is handled by digit feedback
+                self.saved_joint_state.header.stamp = msg.header.stamp
+                self.joint_state_pub_.publish(self.saved_joint_state)
+            case 58:  # REV SPARK MAX position and velocity feedback
+                motorId = round(msg.data[0])
+                motor: RevMotorState | None = None
+                match motorId:
+                    case 1:
+                        motor = self.arm_feedback_new.axis1_motor
+                    case 2:
+                        motor = self.arm_feedback_new.axis2_motor
+                    case 3:
+                        motor = self.arm_feedback_new.axis3_motor
+                    case 4:
+                        motor = self.arm_feedback_new.axis0_motor
+
+                if motor:
+                    motor.position = float(msg.data[1])
+                    motor.velocity = float(msg.data[2])
+                    motor.header.stamp = msg.header.stamp
+
+                self.arm_feedback_pub_.publish(self.arm_feedback_new)
+            case 59:  # Arm joint velocities
+                velocities = [vel / 100.0 for vel in msg.data]  # VicCAN sends deg/s*100
+                self.saved_joint_state.velocity[0] = math.radians(
+                    velocities[0]
+                )  # Axis 0
+                self.saved_joint_state.velocity[1] = math.radians(
+                    velocities[1]
+                )  # Axis 1
+                self.saved_joint_state.velocity[2] = math.radians(
+                    velocities[2]
+                )  # Axis 2
+                self.saved_joint_state.velocity[3] = math.radians(
+                    velocities[3]
+                )  # Axis 3
+                # Wrist is handled by digit feedback
+                self.saved_joint_state.header.stamp = msg.header.stamp
+                self.joint_state_pub_.publish(self.saved_joint_state)
+
+    def process_fromvic_digit(self, msg: VicCAN):
+        assert msg.mcu_name == "digit"
+
+        # Check message len to prevent crashing on bad data
+        if msg.command_id in self.viccan_digit_msg_len_dict:
+            expected_len = self.viccan_digit_msg_len_dict[msg.command_id]
+            if len(msg.data) != expected_len:
+                self.get_logger().warning(
+                    f"Ignoring VicCAN message with id {msg.command_id} due to unexpected data length (expected {expected_len}, got {len(msg.data)})"
+                )
+                return
+
+        match msg.command_id:
+            case 54:  # Board voltages
+                self.arm_feedback_new.digit_voltage.vbatt = float(msg.data[0]) / 100.0
+                self.arm_feedback_new.digit_voltage.v12 = float(msg.data[1]) / 100.0
+                self.arm_feedback_new.digit_voltage.v5 = float(msg.data[2]) / 100.0
+            case 55:  # Arm joint positions
+                self.saved_joint_state.position[4] = math.radians(
+                    msg.data[0]
+                )  # Wrist roll
+                self.saved_joint_state.position[5] = math.radians(
+                    msg.data[1]
+                )  # Wrist yaw
+
+    @deprecated("Uses an old message type. Will be removed at some point.")
     def publish_feedback(self):
         self.socket_pub.publish(self.arm_feedback)
         self.digit_pub.publish(self.digit_feedback)
 
+    @deprecated("Uses an old message type. Will be removed at some point.")
     def updateAngleFeedback(self, msg: str):
         # Angle feedbacks,
         # split the msg.data by commas
@@ -263,19 +410,10 @@ class SerialRelay(Node):
             self.arm_feedback.axis1_angle = angles[1]
             self.arm_feedback.axis2_angle = angles[2]
             self.arm_feedback.axis3_angle = angles[3]
-
-            # Joint state publisher for URDF visualization
-            self.joint_state.position[0] = math.radians(angles[0])  # Axis 0
-            self.joint_state.position[1] = math.radians(angles[1])  # Axis 1
-            self.joint_state.position[2] = math.radians(-angles[2])  # Axis 2 (inverted)
-            self.joint_state.position[3] = math.radians(-angles[3])  # Axis 3 (inverted)
-            # Wrist is handled by digit feedback
-            self.joint_state.header.stamp = self.get_clock().now().to_msg()
-            self.joint_state_pub.publish(self.joint_state)
-
         else:
             self.get_logger().info("Invalid angle feedback input format")
 
+    @deprecated("Uses an old message type. Will be removed at some point.")
     def updateBusVoltage(self, msg: str):
         # Bus Voltage feedbacks
         parts = msg.split(",")
@@ -290,6 +428,7 @@ class SerialRelay(Node):
         else:
             self.get_logger().info("Invalid voltage feedback input format")
 
+    @deprecated("Uses an old message type. Will be removed at some point.")
     def updateMotorFeedback(self, msg: str):
         parts = str(msg.strip()).split(",")
         motorId = round(float(parts[3]))
@@ -313,38 +452,28 @@ class SerialRelay(Node):
             self.arm_feedback.axis0_voltage = voltage
             self.arm_feedback.axis0_current = current
 
-    @staticmethod
-    def list_serial_ports():
-        return glob.glob("/dev/ttyUSB*") + glob.glob("/dev/ttyACM*")
-        # return glob.glob("/dev/tty[A-Za-z]*")
 
-    def cleanup(self):
+def exit_handler(signum, frame):
-        print("Cleaning up...")
+    print("Caught SIGTERM. Exiting...")
-        try:
+    rclpy.try_shutdown()
-            if self.ser.is_open:
+    sys.exit(0)
-                self.ser.close()
-        except Exception as e:
-            exit(0)
-
-
-def myexcepthook(type, value, tb):
-    print("Uncaught exception:", type, value)
-    if serial_pub:
-        serial_pub.cleanup()
 
 
 def main(args=None):
     rclpy.init(args=args)
-    sys.excepthook = myexcepthook
 
-    global serial_pub
+    # Catch termination signals and exit cleanly
-    serial_pub = SerialRelay()
+    signal.signal(signal.SIGTERM, exit_handler)
-    serial_pub.run()
+
+    arm_node = ArmNode()
+
+    try:
+        rclpy.spin(arm_node)
+    except (KeyboardInterrupt, ExternalShutdownException):
+        pass
+    finally:
+        rclpy.try_shutdown()
 
 
 if __name__ == "__main__":
-    # signal.signal(signal.SIGTSTP, lambda signum, frame: sys.exit(0))  # Catch Ctrl+Z and exit cleanly
-    signal.signal(
-        signal.SIGTERM, lambda signum, frame: sys.exit(0)
-    )  # Catch termination signals and exit cleanly
     main()

src/arm_pkg/setup.py

@@ -1,27 +1,22 @@
-from setuptools import find_packages, setup
+from setuptools import setup
-import os
-from glob import glob
 
 package_name = "arm_pkg"
 
 setup(
     name=package_name,
     version="1.0.0",
-    packages=find_packages(exclude=["test"]),
+    packages=[package_name],
     data_files=[
         ("share/ament_index/resource_index/packages", ["resource/" + package_name]),
         ("share/" + package_name, ["package.xml"]),
     ],
     install_requires=["setuptools"],
     zip_safe=True,
-    maintainer="tristan",
+    maintainer="David",
-    maintainer_email="tristanmcginnis26@gmail.com",
+    maintainer_email="ds0196@uah.edu",
-    description="TODO: Package description",
+    description="Relays topics related to the arm between VicCAN (through Anchor) and basestation.",
-    license="All Rights Reserved",
+    license="AGPL-3.0-only",
     entry_points={
-        "console_scripts": [
+        "console_scripts": ["arm = arm_pkg.arm_node:main"],
-            "arm = arm_pkg.arm_node:main",
-            "headless = arm_pkg.arm_headless:main",
-        ],
     },
 )

src/bio_pkg/bio_pkg/bio_node.py

@@ -1,251 +1,241 @@
-import signal
+import rclpy
+from rclpy.node import Node
+import serial
 import sys
 import threading
+import os
+import glob
 import time
-
+import atexit
-import rclpy
+import signal
-from astra_msgs.action import BioVacuum
+from std_msgs.msg import String
-from astra_msgs.msg import CitadelControl, FaerieControl, VicCAN
+from astra_msgs.msg import BioControl
-from astra_msgs.srv import BioTestTube, LibsSystem
+from astra_msgs.msg import BioFeedback
-from rclpy.action import ActionServer
-from rclpy.node import Node
-from std_msgs.msg import Header, String
 
 serial_pub = None
 thread = None
 
-# used to verify the length of an incoming VicCAN feedback message
-# key is VicCAN command_id, value is expected length of data list
-viccan_citadel_msg_len_dict = {
-    # empty because not expecting any VicCAN from citadel atm
-}
-viccan_lance_msg_len_dict = {
-    # empty because not sure what VicCAN commands LANCE sends
-}
-
 
 class SerialRelay(Node):
     def __init__(self):
         # Initialize node
         super().__init__("bio_node")
 
-        # Anchor Topics
+        # Get launch mode parameter
-        self.anchor_fromvic_sub_ = self.create_subscription(
+        self.declare_parameter("launch_mode", "bio")
-            VicCAN, "/anchor/from_vic/bio", self.relay_fromvic, 20
+        self.launch_mode = self.get_parameter("launch_mode").value
-        )
+        self.get_logger().info(f"bio launch_mode is: {self.launch_mode}")
-        self.anchor_tovic_pub_ = self.create_publisher(
+
-            VicCAN, "/anchor/to_vic/relay", 20
+        # Create publishers
+        self.debug_pub = self.create_publisher(String, "/bio/feedback/debug", 10)
+        self.feedback_pub = self.create_publisher(BioFeedback, "/bio/feedback", 10)
+
+        # Create subscribers
+        self.control_sub = self.create_subscription(
+            BioControl, "/bio/control", self.send_control, 10
         )
 
+        # Create a publisher for telemetry
+        self.telemetry_pub_timer = self.create_timer(1.0, self.publish_feedback)
+
+        # Topics used in anchor mode
+        if self.launch_mode == "anchor":
             self.anchor_sub = self.create_subscription(
                 String, "/anchor/bio/feedback", self.anchor_feedback, 10
             )
             self.anchor_pub = self.create_publisher(String, "/anchor/relay", 10)
 
-        # Messages
+        self.bio_feedback = BioFeedback()
-        self.citadel_sub = self.create_subscription(
-            CitadelControl,
-            "/bio/control/citadel",
-            self.citadel_callback,
-            10,
-        )
 
-        self.faerie_sub = self.create_subscription(
+        # Search for ports IF in 'arm' (standalone) and not 'anchor' mode
-            FaerieControl,
+        if self.launch_mode == "bio":
-            "/bio/control/faerie",
+            # Loop through all serial devices on the computer to check for the MCU
-            self.faerie_callback,
+            self.port = None
-            10,
+            for i in range(2):
+                try:
+                    # connect and send a ping command
+                    set_port = (
+                        "/dev/ttyACM0"  # MCU is controlled through GPIO pins on the PI
                     )
+                    ser = serial.Serial(set_port, 115200, timeout=1)
+                    # print(f"Checking port {port}...")
+                    ser.write(b"ping\n")
+                    response = ser.read_until("\n")
 
-        # Services
+                    # if pong is in response, then we are talking with the MCU
-        self.test_tube_service = self.create_service(
+                    if b"pong" in response:
-            BioTestTube, "/bio/control/test_tube", self.test_tube_callback
+                        self.port = set_port
-        )
+                        self.get_logger().info(f"Found MCU at {set_port}!")
-        self.libs_service = self.create_service(
+                        break
-            LibsSystem, "bio/control/libs_system", self.libs_callback
+                except:
-        )
+                    pass
 
-        # Actions
+            if self.port is None:
-        self._action_server = ActionServer(
+                self.get_logger().info(
-            self, BioVacuum, "/bio/actions/vacuum", self.execute_vacuum
+                    "Unable to find MCU... please make sure it is connected."
                 )
+                time.sleep(1)
+                sys.exit(1)
+
+            self.ser = serial.Serial(self.port, 115200)
+            atexit.register(self.cleanup)
 
     def run(self):
         global thread
         thread = threading.Thread(target=rclpy.spin, args=(self,), daemon=True)
         thread.start()
 
+        # if in arm mode, will need to read from the MCU
+
         try:
             while rclpy.ok():
+                if self.launch_mode == "bio":
+                    if self.ser.in_waiting:
+                        self.read_mcu()
+                    else:
                         time.sleep(0.1)
         except KeyboardInterrupt:
             pass
+        finally:
+            self.cleanup()
+
+    # Currently will just spit out all values over the /arm/feedback/debug topic as strings
+    def read_mcu(self):
+        try:
+            output = str(self.ser.readline(), "utf8")
+            if output:
+                self.get_logger().info(f"[MCU] {output}")
+                msg = String()
+                msg.data = output
+                self.debug_pub.publish(msg)
+        except serial.SerialException:
+            self.get_logger().info("SerialException caught... closing serial port.")
+            if self.ser.is_open:
+                self.ser.close()
+            pass
+        except TypeError as e:
+            self.get_logger().info(f"TypeError: {e}")
+            print("Closing serial port.")
+            if self.ser.is_open:
+                self.ser.close()
+            pass
+        except Exception as e:
+            print(f"Exception: {e}")
+            print("Closing serial port.")
+            if self.ser.is_open:
+                self.ser.close()
+            pass
+
+    def send_ik(self, msg):
+        pass
+
+    def send_control(self, msg: BioControl):
+        # CITADEL Control Commands
+        ################
+
+        # Chem Pumps, only send if not zero
+        if msg.pump_id != 0:
+            command = (
+                "can_relay_tovic,citadel,27,"
+                + str(msg.pump_id)
+                + ","
+                + str(msg.pump_amount)
+                + "\n"
+            )
+            self.send_cmd(command)
+        # Fans, only send if not zero
+        if msg.fan_id != 0:
+            command = (
+                "can_relay_tovic,citadel,40,"
+                + str(msg.fan_id)
+                + ","
+                + str(msg.fan_duration)
+                + "\n"
+            )
+            self.send_cmd(command)
+        # Servos, only send if not zero
+        if msg.servo_id != 0:
+            command = (
+                "can_relay_tovic,citadel,25,"
+                + str(msg.servo_id)
+                + ","
+                + str(int(msg.servo_state))
+                + "\n"
+            )
+            self.send_cmd(command)
+
+        # LSS (SCYTHE)
+        command = "can_relay_tovic,citadel,24," + str(msg.bio_arm) + "\n"
+        # self.send_cmd(command)
+
+        # Vibration Motor
+        command += "can_relay_tovic,citadel,26," + str(msg.vibration_motor) + "\n"
+        # self.send_cmd(command)
+
+        # FAERIE Control Commands
+        ################
+
+        # To be reviewed before use#
+
+        # Laser
+        command += "can_relay_tovic,digit,28," + str(msg.laser) + "\n"
+        # self.send_cmd(command)
+
+        # Drill (SCABBARD)
+        command += f"can_relay_tovic,digit,19,{msg.drill:.2f}\n"
+        # self.send_cmd(command)
+
+        # Bio linear actuator
+        command += "can_relay_tovic,digit,42," + str(msg.drill_arm) + "\n"
+        self.send_cmd(command)
 
     def send_cmd(self, msg: str):
-        # send to anchor node to relay
+        if (
+            self.launch_mode == "anchor"
+        ):  # if in anchor mode, send to anchor node to relay
             output = String()
             output.data = msg
             self.anchor_pub.publish(output)
-
+        elif self.launch_mode == "bio":  # if in standalone mode, send to MCU directly
-    def relay_fromvic(self, msg: VicCAN):
+            self.get_logger().info(f"[Bio to MCU] {msg}")
-        # self.get_logger().info(msg)
+            self.ser.write(bytes(msg, "utf8"))
-        if msg.mcu_name == "citadel":
-            self.process_fromvic_citadel(msg)
-
-    def process_fromvic_citadel(self, msg: VicCAN):
-        # Check message len to prevent crashing on bad data
-        if msg.command_id in viccan_citadel_msg_len_dict:
-            expected_len = viccan_citadel_msg_len_dict[msg.command_id]
-            if len(msg.data) != expected_len:
-                self.get_logger().warning(
-                    f"Ignoring VicCAN message with id {msg.command_id} due to unexpected data length (expected {expected_len}, got {len(msg.data)})"
-                )
-            return
 
     def anchor_feedback(self, msg: String):
         output = msg.data
         parts = str(output.strip()).split(",")
-        self.get_logger().info(f"[Bio Anchor] {msg.data}")
+        # self.get_logger().info(f"[Bio Anchor] {msg.data}")
-        # no data planned to be received from citadel, not sure about lance
 
-    def citadel_callback(self, msg: CitadelControl):
+        if output.startswith(
-        distributor_arr = msg.distributor_id
+            "can_relay_fromvic,citadel,54"
-        # Distributor Control
+        ):  # bat, 12, 5, Voltage readings * 100
-        vic_cmd = VicCAN(
+            self.bio_feedback.bat_voltage = float(parts[3]) / 100.0
-            header=Header(stamp=self.get_clock().now().to_msg()),
+            self.bio_feedback.voltage_12 = float(parts[4]) / 100.0
-            mcu_name="citadel",
+            self.bio_feedback.voltage_5 = float(parts[5]) / 100.0
-            command_id=40,
+        elif output.startswith("can_relay_fromvic,digit,57"):
-            data=[
+            self.bio_feedback.drill_temp = float(parts[3])
-                clamp_short(distributor_arr[0]),
+            self.bio_feedback.drill_humidity = float(parts[4])
-                clamp_short(distributor_arr[1]),
-                clamp_short(distributor_arr[2]),
-                0,
-            ],
-        )
-        self.anchor_tovic_pub_.publish(vic_cmd)
-        # Move Scythe
-        vic_cmd = VicCAN(
-            header=Header(stamp=self.get_clock().now().to_msg()),
-            mcu_name="digit",
-            command_id=42,
-            data=[float(msg.move_scythe)],
-        )
-        self.anchor_tovic_pub_.publish(vic_cmd)
 
-    def faerie_callback(self, msg: FaerieControl):
+    def publish_feedback(self):
-        # Move Faerie
+        self.feedback_pub.publish(self.bio_feedback)
-        vic_cmd = VicCAN(
-            header=Header(stamp=self.get_clock().now().to_msg()),
-            mcu_name="digit",
-            command_id=42,
-            data=[float(msg.move_faerie)],
-        )
-        self.anchor_tovic_pub_.publish(vic_cmd)
-        # Drill Speed
-        vic_cmd = VicCAN(
-            header=Header(stamp=self.get_clock().now().to_msg()),
-            mcu_name="digit",
-            command_id=19,
-            data=[
-                float(msg.drill_speed * 100)
-            ],  # change on embedded so we can go (-1,1)
-        )
-        self.anchor_tovic_pub_.publish(vic_cmd)
-        # Vanity Laser Control
-        vic_cmd = VicCAN(
-            header=Header(stamp=self.get_clock().now().to_msg()),
-            mcu_name="digit",
-            command_id=28,
-            data=[float(msg.vanity_laser)],
-        )
-        self.anchor_tovic_pub_.publish(vic_cmd)
 
-    def test_tube_callback(self, request, response):
+    @staticmethod
-        vic_cmd = VicCAN(
+    def list_serial_ports():
-            header=Header(stamp=self.get_clock().now().to_msg()),
+        return glob.glob("/dev/ttyUSB*") + glob.glob("/dev/ttyACM*")
-            mcu_name="citadel",
+        # return glob.glob("/dev/tty[A-Za-z]*")
-            command_id=40,
-            data=[
-                float(int(request.tube_id)),
-                float(request.milliliters),
-            ],
-        )
-        self.anchor_tovic_pub_.publish(vic_cmd)
-        return response
 
-    def libs_callback(self, request, response):
+    def cleanup(self):
-        print("todo")
+        print("Cleaning up...")
-
+        try:
-    def execute_vacuum(self, goal_handle):
+            if self.ser.is_open:
-        valve_id = int(goal_handle.request.valve_id)
+                self.ser.close()
-        duty = int(goal_handle.request.fan_duty_cycle_percent)
+        except Exception as e:
-        total = goal_handle.request.fan_time_ms
+            exit(0)
-
-        # open valve
-        self.anchor_tovic_pub_.publish(
-            VicCAN(
-                header=Header(stamp=self.get_clock().now().to_msg()),
-                mcu_name="citadel",
-                command_id=40,
-                data=[float(valve_id)],
-            )
-        )
-
-        feedback = BioVacuum.Feedback()
-        start = time.time()
-
-        while True:
-            # set fan duty cycle
-            self.anchor_tovic_pub_.publish(
-                VicCAN(
-                    header=Header(stamp=self.get_clock().now().to_msg()),
-                    mcu_name="citadel",
-                    command_id=19,
-                    data=[float(duty)],
-                )
-            )
-
-            elapsed = int((time.time() - start) * 1000)
-            remaining = max(0, total - elapsed)
-
-            feedback.fan_time_remaining_ms = remaining
-            goal_handle.publish_feedback(feedback)
-
-            if remaining == 0:
-                break
-
-            time.sleep(0.1)
-
-        # stop fan
-        self.anchor_tovic_pub_.publish(
-            VicCAN(
-                header=Header(stamp=self.get_clock().now().to_msg()),
-                mcu_name="citadel",
-                command_id=19,
-                data=[0.0],
-            )
-        )
-
-        # close valve
-        self.anchor_tovic_pub_.publish(
-            VicCAN(
-                header=Header(stamp=self.get_clock().now().to_msg()),
-                mcu_name="citadel",
-                command_id=40,
-                data=[-1.0],
-            )
-        )
-
-        goal_handle.succeed()
-        return BioVacuum.Result()
-
-
-def clamp_short(x: int) -> int:
-    return max(-32768, min(32767, x))
 
 
 def myexcepthook(type, value, tb):
     print("Uncaught exception:", type, value)
+    if serial_pub:
+        serial_pub.cleanup()
 
 
 def main(args=None):

src/core_pkg/core_pkg/core_node.py

@@ -31,10 +31,10 @@ CORE_WHEEL_RADIUS = 0.171  # meters
 CORE_GEAR_RATIO = 100.0  # Clucky: 100:1, Testbed: 64:1
 
 control_qos = qos.QoSProfile(
-    # history=qos.QoSHistoryPolicy.KEEP_LAST,
+    history=qos.QoSHistoryPolicy.KEEP_LAST,
     depth=2,
-    # reliability=qos.QoSReliabilityPolicy.BEST_EFFORT,
+    reliability=qos.QoSReliabilityPolicy.BEST_EFFORT,  # Best Effort subscribers are still compatible with Reliable publishers
-    # durability=qos.QoSDurabilityPolicy.VOLATILE,
+    durability=qos.QoSDurabilityPolicy.VOLATILE,
     # deadline=Duration(seconds=1),
     # lifespan=Duration(nanoseconds=500_000_000),  # 500ms
     # liveliness=qos.QoSLivelinessPolicy.SYSTEM_DEFAULT,

src/headless_pkg/src/headless_node.py

@@ -6,21 +6,28 @@ from rclpy.duration import Duration
 
 import signal
 import time
-import atexit
 
 import os
 import sys
-import threading
-import glob
 import pwd
 import grp
 from math import copysign
 
-from std_msgs.msg import String
+from std_srvs.srv import Trigger
-from geometry_msgs.msg import Twist
+from std_msgs.msg import Header
+from geometry_msgs.msg import Twist, TwistStamped
+from control_msgs.msg import JointJog
 from astra_msgs.msg import CoreControl, ArmManual, BioControl
 from astra_msgs.msg import CoreCtrlState
 
+import warnings
+
+# Literally headless
+warnings.filterwarnings(
+    "ignore",
+    message="Your system is avx2 capable but pygame was not built with support for it.",
+)
+
 import pygame
 
 os.environ["SDL_VIDEODRIVER"] = "dummy"  # Prevents pygame from trying to open a display
@@ -34,26 +41,44 @@ CORE_STOP_TWIST_MSG = Twist()  # "
 ARM_STOP_MSG = ArmManual()  # "
 BIO_STOP_MSG = BioControl()  # "
 
+
 control_qos = qos.QoSProfile(
-    # history=qos.QoSHistoryPolicy.KEEP_LAST,
+    history=qos.QoSHistoryPolicy.KEEP_LAST,
     depth=2,
-    # reliability=qos.QoSReliabilityPolicy.BEST_EFFORT,
+    reliability=qos.QoSReliabilityPolicy.BEST_EFFORT,
-    # durability=qos.QoSDurabilityPolicy.VOLATILE,
+    durability=qos.QoSDurabilityPolicy.VOLATILE,
     # deadline=Duration(seconds=1),
     # lifespan=Duration(nanoseconds=500_000_000),  # 500ms
     # liveliness=qos.QoSLivelinessPolicy.SYSTEM_DEFAULT,
     # liveliness_lease_duration=Duration(seconds=5),
 )
 
-CORE_MODE = "twist"  # "twist" or "duty"
+
+STICK_DEADZONE = float(os.getenv("STICK_DEADZONE", "0.05"))
+ARM_DEADZONE = float(os.getenv("ARM_DEADZONE", "0.2"))
 
 
 class Headless(Node):
+    # Every non-fixed joint defined in Arm's URDF
+    # Used for JointState and JointJog messsages
+    all_joint_names = [
+        "axis_0_joint",
+        "axis_1_joint",
+        "axis_2_joint",
+        "axis_3_joint",
+        "wrist_yaw_joint",
+        "wrist_roll_joint",
+        "ef_gripper_left_joint",
+    ]
+
     def __init__(self):
         # Initialize pygame first
         pygame.init()
         pygame.joystick.init()
-        super().__init__("headless")
+        super().__init__("headless_node")
+
+        ##################################################
+        # Preamble
 
         # Wait for anchor to start
         pub_info = self.get_publishers_info_by_topic("/anchor/from_vic/debug")
@@ -105,12 +130,57 @@ class Headless(Node):
                 break
             id += 1
 
-        self.create_timer(0.15, self.send_controls)
+        ##################################################
+        # Parameters
 
+        self.declare_parameter("use_old_topics", True)
+        self.use_old_topics = (
+            self.get_parameter("use_old_topics").get_parameter_value().bool_value
+        )
+
+        self.declare_parameter("use_bio", False)
+        self.use_bio = self.get_parameter("use_bio").get_parameter_value().bool_value
+
+        self.declare_parameter("use_arm_ik", False)
+        self.use_arm_ik = (
+            self.get_parameter("use_arm_ik").get_parameter_value().bool_value
+        )
+
+        # NOTE: only applicable if use_old_topics == True
+        self.declare_parameter("use_new_arm_manual_scheme", True)
+        self.use_new_arm_manual_scheme = (
+            self.get_parameter("use_new_arm_manual_scheme")
+            .get_parameter_value()
+            .bool_value
+        )
+
+        # Check parameter validity
+        if self.use_arm_ik and self.use_old_topics:
+            self.get_logger().fatal("Old topics do not support arm IK control.")
+            sys.exit(1)
+        if not self.use_new_arm_manual_scheme and not self.use_old_topics:
+            self.get_logger().warn(
+                f"New arm manual does not support old control scheme. Defaulting to new scheme."
+            )
+
+        self.ctrl_mode = "core"  # Start in core mode
+        self.core_brake_mode = False
+        self.core_max_duty = 0.5  # Default max duty cycle (walking speed)
+
+        ##################################################
+        # Old Topics
+
+        if self.use_old_topics:
             self.core_publisher = self.create_publisher(CoreControl, "/core/control", 2)
-        self.arm_publisher = self.create_publisher(ArmManual, "/arm/control/manual", 2)
+            self.arm_publisher = self.create_publisher(
+                ArmManual, "/arm/control/manual", 2
+            )
             self.bio_publisher = self.create_publisher(BioControl, "/bio/control", 2)
 
+        ##################################################
+        # New Topics
+
+        if not self.use_old_topics:
             self.core_twist_pub_ = self.create_publisher(
                 Twist, "/core/twist", qos_profile=control_qos
             )
@@ -118,24 +188,61 @@ class Headless(Node):
                 CoreCtrlState, "/core/control/state", qos_profile=control_qos
             )
 
-        self.ctrl_mode = "core"  # Start in core mode
+            self.arm_manual_pub_ = self.create_publisher(
-        self.core_brake_mode = False
+                JointJog, "/arm/manual_new", qos_profile=control_qos
-        self.core_max_duty = 0.5  # Default max duty cycle (walking speed)
+            )
+
+            self.arm_ik_twist_publisher = self.create_publisher(
+                TwistStamped, "/servo_node/delta_twist_cmds", qos_profile=control_qos
+            )
+            self.arm_ik_jointjog_publisher = self.create_publisher(
+                JointJog, "/servo_node/delta_joint_cmds", qos_profile=control_qos
+            )
+
+            # TODO: add new bio topics
+
+        ##################################################
+        # Timers
+
+        self.create_timer(0.1, self.send_controls)
+
+        ##################################################
+        # Services
+
+        # If using IK control, we have to "start" the servo node to enable it to accept commands
+        self.servo_start_client = None
+        if self.use_arm_ik:
+            self.get_logger().info("Starting servo node for IK control...")
+            self.servo_start_client = self.create_client(
+                Trigger, "/servo_node/start_servo"
+            )
+            timeout_counter = 0
+            while not self.servo_start_client.wait_for_service(timeout_sec=1.0):
+                self.get_logger().info("Waiting for servo_node/start_servo service...")
+                timeout_counter += 1
+                if timeout_counter >= 10:
+                    self.get_logger().error(
+                        "Servo's start service not available. IK control will not work."
+                    )
+                    break
+            if self.servo_start_client.service_is_ready():
+                self.servo_start_client.call_async(Trigger.Request())
 
         # Rumble when node is ready (returns False if rumble not supported)
         self.gamepad.rumble(0.7, 0.8, 150)
 
-    def run(self):
+    def stop_all(self):
-        # This thread makes all the update processes run in the background
+        if self.use_old_topics:
-        thread = threading.Thread(target=rclpy.spin, args={self}, daemon=True)
+            self.core_publisher.publish(CORE_STOP_MSG)
-        thread.start()
+            self.arm_publisher.publish(ARM_STOP_MSG)
-
+            self.bio_publisher.publish(BIO_STOP_MSG)
-        try:
+        else:
-            while rclpy.ok():
+            self.core_twist_pub_.publish(CORE_STOP_TWIST_MSG)
-                self.send_controls()
+            if self.use_arm_ik:
-                time.sleep(0.1)  # Small delay to avoid CPU hogging
+                self.arm_ik_twist_publisher.publish(self.arm_ik_twist_stop_msg())
-        except KeyboardInterrupt:
+            else:
-            sys.exit(0)
+                self.arm_manual_pub_.publish(self.arm_manual_stop_msg())
+            # TODO: add bio here after implementing new topics
 
     def send_controls(self):
         """Read the gamepad state and publish control messages"""
@@ -147,10 +254,8 @@ class Headless(Node):
         # Check if controller is still connected
         if pygame.joystick.get_count() != self.num_gamepads:
             print("Gamepad disconnected. Exiting...")
-            # Send one last zero control message
+            # Stop the rover if controller disconnected
-            self.core_publisher.publish(CORE_STOP_MSG)
+            self.stop_all()
-            self.arm_publisher.publish(ARM_STOP_MSG)
-            self.bio_publisher.publish(BIO_STOP_MSG)
             self.get_logger().info("Final stop commands sent. Shutting down.")
             # Clean up
             pygame.quit()
@@ -166,20 +271,51 @@ class Headless(Node):
             new_ctrl_mode = "core"
 
         if new_ctrl_mode != self.ctrl_mode:
+            self.stop_all()
             self.gamepad.rumble(0.6, 0.7, 75)
             self.ctrl_mode = new_ctrl_mode
             self.get_logger().info(f"Switched to {self.ctrl_mode} control mode")
+            if self.ctrl_mode == "arm" and self.use_bio:
+                self.get_logger().warning("NOTE: Using bio instead of arm.")
 
-        # CORE
+        # Actually send the controls
-        if self.ctrl_mode == "core" and CORE_MODE == "duty":
+        if self.ctrl_mode == "core":
+            self.send_core()
+            if self.use_old_topics:
+                if self.use_bio:
+                    self.bio_publisher.publish(BIO_STOP_MSG)
+                else:
+                    self.arm_publisher.publish(ARM_STOP_MSG)
+            # New topics shouldn't need to constantly send zeroes imo
+        else:
+            if self.use_bio:
+                self.send_bio()
+            else:
+                self.send_arm()
+            if self.use_old_topics:
+                self.core_publisher.publish(CORE_STOP_MSG)
+            # Ditto
+
+    def send_core(self):
+        # Collect controller state
+        left_stick_x = stick_deadzone(self.gamepad.get_axis(0))
+        left_stick_y = stick_deadzone(self.gamepad.get_axis(1))
+        left_trigger = stick_deadzone(self.gamepad.get_axis(2))
+        right_stick_x = stick_deadzone(self.gamepad.get_axis(3))
+        right_stick_y = stick_deadzone(self.gamepad.get_axis(4))
+        right_trigger = stick_deadzone(self.gamepad.get_axis(5))
+        button_a = self.gamepad.get_button(0)
+        button_b = self.gamepad.get_button(1)
+        button_x = self.gamepad.get_button(2)
+        button_y = self.gamepad.get_button(3)
+        left_bumper = self.gamepad.get_button(4)
+        right_bumper = self.gamepad.get_button(5)
+        dpad_input = self.gamepad.get_hat(0)
+
+        if self.use_old_topics:
             input = CoreControl()
             input.max_speed = 90
 
-            # Collect controller state
-            left_stick_y = deadzone(self.gamepad.get_axis(1))
-            right_stick_y = deadzone(self.gamepad.get_axis(4))
-            right_trigger = deadzone(self.gamepad.get_axis(5))
-
             # Right wheels
             input.right_stick = float(round(-1 * right_stick_y, 2))
 
@@ -194,19 +330,10 @@ class Headless(Node):
             self.get_logger().info(f"[Ctrl] {output}")
 
             self.core_publisher.publish(input)
-            self.arm_publisher.publish(ARM_STOP_MSG)
-            # self.bio_publisher.publish(BIO_STOP_MSG)
 
-        elif self.ctrl_mode == "core" and CORE_MODE == "twist":
+        else:  # New topics
             input = Twist()
 
-            # Collect controller state
-            left_stick_y = deadzone(self.gamepad.get_axis(1))
-            right_stick_x = deadzone(self.gamepad.get_axis(3))
-            button_a = self.gamepad.get_button(0)
-            left_bumper = self.gamepad.get_button(4)
-            right_bumper = self.gamepad.get_button(5)
-
             # Forward/back and Turn
             input.linear.x = -1.0 * left_stick_y
             input.angular.z = -1.0 * copysign(
@@ -215,8 +342,6 @@ class Headless(Node):
 
             # Publish
             self.core_twist_pub_.publish(input)
-            self.arm_publisher.publish(ARM_STOP_MSG)
-            # self.bio_publisher.publish(BIO_STOP_MSG)
             self.get_logger().info(
                 f"[Core Ctrl] Linear: {round(input.linear.x, 2)}, Angular: {round(input.angular.z, 2)}"
             )
@@ -241,25 +366,36 @@ class Headless(Node):
                 state_msg = CoreCtrlState()
                 state_msg.brake_mode = bool(self.core_brake_mode)
                 state_msg.max_duty = float(self.core_max_duty)
+
                 self.core_state_pub_.publish(state_msg)
                 self.get_logger().info(
                     f"[Core State] Brake: {self.core_brake_mode}, Max Duty: {self.core_max_duty}"
                 )
 
-        # ARM and BIO
+    def send_arm(self):
-        if self.ctrl_mode == "arm":
-            arm_input = ArmManual()
-
         # Collect controller state
-            left_stick_x = deadzone(self.gamepad.get_axis(0))
+        left_stick_x = stick_deadzone(self.gamepad.get_axis(0))
-            left_stick_y = deadzone(self.gamepad.get_axis(1))
+        left_stick_y = stick_deadzone(self.gamepad.get_axis(1))
-            left_trigger = deadzone(self.gamepad.get_axis(2))
+        left_trigger = stick_deadzone(self.gamepad.get_axis(2))
-            right_stick_x = deadzone(self.gamepad.get_axis(3))
+        right_stick_x = stick_deadzone(self.gamepad.get_axis(3))
-            right_stick_y = deadzone(self.gamepad.get_axis(4))
+        right_stick_y = stick_deadzone(self.gamepad.get_axis(4))
-            right_trigger = deadzone(self.gamepad.get_axis(5))
+        right_trigger = stick_deadzone(self.gamepad.get_axis(5))
+        button_a = self.gamepad.get_button(0)
+        button_b = self.gamepad.get_button(1)
+        button_x = self.gamepad.get_button(2)
+        button_y = self.gamepad.get_button(3)
+        left_bumper = self.gamepad.get_button(4)
         right_bumper = self.gamepad.get_button(5)
         dpad_input = self.gamepad.get_hat(0)
 
+        # OLD MANUAL
+        # ==========
+
+        if not self.use_arm_ik and self.use_old_topics:
+            arm_input = ArmManual()
+
+            # OLD ARM MANUAL CONTROL SCHEME
+            if not self.use_new_arm_manual_scheme:
                 # EF Grippers
                 if left_trigger > 0 and right_trigger > 0:
                     arm_input.gripper = 0
@@ -302,30 +438,238 @@ class Headless(Node):
                     if abs(right_stick_y) > 0.15:
                         arm_input.axis3 = -1 * round(right_stick_y)
 
-            # BIO
+            # NEW ARM MANUAL CONTROL SCHEME
+            if self.use_new_arm_manual_scheme:
+                # Right stick: EF yaw and axis 3
+                # Left stick: axis 1 and 2
+                # D-pad: axis 0 and _
+                # Triggers: EF grippers
+                # Bumpers: EF roll
+                # A: brake
+                # B: linear actuator in
+                # X: _
+                # Y: linear actuator out
+
+                # Right stick: EF yaw and axis 3
+                arm_input.effector_yaw = stick_to_arm_direction(right_stick_x)
+                arm_input.axis3 = -1 * stick_to_arm_direction(right_stick_y)
+
+                # Left stick: axis 1 and 2
+                arm_input.axis1 = stick_to_arm_direction(left_stick_x)
+                arm_input.axis2 = -1 * stick_to_arm_direction(left_stick_y)
+
+                # D-pad: axis 0 and _
+                arm_input.axis0 = int(dpad_input[0])
+
+                # Triggers: EF Grippers
+                if left_trigger > 0 and right_trigger > 0:
+                    arm_input.gripper = 0
+                elif left_trigger > 0:
+                    arm_input.gripper = -1
+                elif right_trigger > 0:
+                    arm_input.gripper = 1
+
+                # Bumpers: EF roll
+                if left_bumper > 0 and right_bumper > 0:
+                    arm_input.effector_roll = 0
+                elif left_bumper > 0:
+                    arm_input.effector_roll = -1
+                elif right_bumper > 0:
+                    arm_input.effector_roll = 1
+
+                # A: brake
+                if button_a:
+                    arm_input.brake = True
+
+                # Y: linear actuator
+                if button_y and not button_b:
+                    arm_input.linear_actuator = 1
+                elif button_b and not button_y:
+                    arm_input.linear_actuator = -1
+                else:
+                    arm_input.linear_actuator = 0
+
+            self.arm_publisher.publish(arm_input)
+
+        # NEW MANUAL
+        # ==========
+
+        elif not self.use_arm_ik and not self.use_old_topics:
+            arm_input = JointJog()
+            arm_input.header.frame_id = "base_link"
+            arm_input.header.stamp = self.get_clock().now().to_msg()
+            arm_input.joint_names = self.all_joint_names
+            arm_input.velocities = [0.0] * len(self.all_joint_names)
+
+            # Right stick: EF yaw and axis 3
+            # Left stick: axis 1 and 2
+            # D-pad: axis 0 and _
+            # Triggers: EF grippers
+            # Bumpers: EF roll
+            # A: brake
+            # B: linear actuator in
+            # X: _
+            # Y: linear actuator out
+
+            # Right stick: EF yaw and axis 3
+            arm_input.velocities[self.all_joint_names.index("wrist_yaw_joint")] = float(
+                stick_to_arm_direction(right_stick_x)
+            )
+            arm_input.velocities[self.all_joint_names.index("axis_3_joint")] = float(
+                stick_to_arm_direction(right_stick_y)
+            )
+
+            # Left stick: axis 1 and 2
+            arm_input.velocities[self.all_joint_names.index("axis_1_joint")] = float(
+                stick_to_arm_direction(left_stick_x)
+            )
+            arm_input.velocities[self.all_joint_names.index("axis_2_joint")] = float(
+                stick_to_arm_direction(left_stick_y)
+            )
+
+            # D-pad: axis 0 and _
+            arm_input.velocities[self.all_joint_names.index("axis_0_joint")] = float(
+                dpad_input[0]
+            )
+
+            # Triggers: EF Grippers
+            if left_trigger > 0 and right_trigger > 0:
+                arm_input.velocities[
+                    self.all_joint_names.index("ef_gripper_left_joint")
+                ] = 0.0
+            elif left_trigger > 0:
+                arm_input.velocities[
+                    self.all_joint_names.index("ef_gripper_left_joint")
+                ] = -1.0
+            elif right_trigger > 0:
+                arm_input.velocities[
+                    self.all_joint_names.index("ef_gripper_left_joint")
+                ] = 1.0
+
+            # Bumpers: EF roll
+            arm_input.velocities[self.all_joint_names.index("wrist_roll_joint")] = (
+                right_bumper - left_bumper
+            )
+
+            # A: brake
+            # TODO: Brake mode
+
+            # Y: linear actuator
+            # TODO: linear actuator
+
+            self.arm_manual_pub_.publish(arm_input)
+
+        # IK (ONLY NEW)
+        # =============
+
+        elif self.use_arm_ik:
+            arm_twist = TwistStamped()
+            arm_twist.header.frame_id = "base_link"
+            arm_twist.header.stamp = self.get_clock().now().to_msg()
+            arm_jointjog = JointJog()
+            arm_jointjog.header.frame_id = "base_link"
+            arm_jointjog.header.stamp = self.get_clock().now().to_msg()
+
+            # Right stick: linear y and linear x
+            # Left stick: angular z and linear z
+            # D-pad: angular y and _
+            # Triggers: EF grippers
+            # Bumpers: angular x
+            # A: brake
+            # B: IK mode
+            # X: manual mode
+            # Y: linear actuator
+
+            # Right stick: linear y and linear x
+            arm_twist.twist.linear.y = float(right_stick_x)
+            arm_twist.twist.linear.x = float(right_stick_y)
+
+            # Left stick: angular z and linear z
+            arm_twist.twist.angular.z = float(-1 * left_stick_x)
+            arm_twist.twist.linear.z = float(-1 * left_stick_y)
+            # D-pad: angular y and _
+            arm_twist.twist.angular.y = (
+                float(0)
+                if dpad_input[0] == 0
+                else float(-1 * copysign(0.75, dpad_input[0]))
+            )
+
+            # Triggers: EF Grippers
+            if left_trigger > 0 or right_trigger > 0:
+                arm_jointjog.joint_names.append("ef_gripper_left_joint")  # type: ignore
+                arm_jointjog.velocities.append(float(right_trigger - left_trigger))
+
+            # Bumpers: angular x
+            if left_bumper > 0 and right_bumper > 0:
+                arm_twist.twist.angular.x = float(0)
+            elif left_bumper > 0:
+                arm_twist.twist.angular.x = float(1)
+            elif right_bumper > 0:
+                arm_twist.twist.angular.x = float(-1)
+
+            self.arm_ik_twist_publisher.publish(arm_twist)
+            # self.arm_ik_jointjog_publisher.publish(arm_jointjog)  # TODO: Figure this shit out
+
+    def send_bio(self):
+        # Collect controller state
+        left_stick_x = stick_deadzone(self.gamepad.get_axis(0))
+        left_stick_y = stick_deadzone(self.gamepad.get_axis(1))
+        left_trigger = stick_deadzone(self.gamepad.get_axis(2))
+        right_stick_x = stick_deadzone(self.gamepad.get_axis(3))
+        right_stick_y = stick_deadzone(self.gamepad.get_axis(4))
+        right_trigger = stick_deadzone(self.gamepad.get_axis(5))
+        button_a = self.gamepad.get_button(0)
+        button_b = self.gamepad.get_button(1)
+        button_x = self.gamepad.get_button(2)
+        button_y = self.gamepad.get_button(3)
+        left_bumper = self.gamepad.get_button(4)
+        right_bumper = self.gamepad.get_button(5)
+        dpad_input = self.gamepad.get_hat(0)
+
+        if self.use_old_topics:
             bio_input = BioControl(
                 bio_arm=int(left_stick_y * -100),
                 drill_arm=int(round(right_stick_y) * -100),
             )
 
             # Drill motor (FAERIE)
-            if deadzone(left_trigger) > 0 or deadzone(right_trigger) > 0:
+            if left_trigger > 0 or right_trigger > 0:
                 bio_input.drill = int(
                     30 * (right_trigger - left_trigger)
                 )  # Max duty cycle 30%
 
-            self.core_publisher.publish(CORE_STOP_MSG)
+            self.bio_publisher.publish(bio_input)
-            self.arm_publisher.publish(arm_input)
+
-            # self.bio_publisher.publish(bio_input)
+        else:
+            pass  # TODO: implement new bio control topics
+
+    def arm_manual_stop_msg(self):
+        return JointJog(
+            header=Header(frame_id="base_link", stamp=self.get_clock().now().to_msg()),
+            joint_names=self.all_joint_names,
+            velocities=[0.0] * len(self.all_joint_names),
+        )
+
+    def arm_ik_twist_stop_msg(self):
+        return TwistStamped(
+            header=Header(frame_id="base_link", stamp=self.get_clock().now().to_msg())
+        )
 
 
-def deadzone(value: float, threshold=0.05) -> float:
+def stick_deadzone(value: float, threshold=STICK_DEADZONE) -> float:
     """Apply a deadzone to a joystick input so the motors don't sound angry"""
     if abs(value) < threshold:
         return 0
     return value
 
 
+def stick_to_arm_direction(value: float, threshold=ARM_DEADZONE) -> int:
+    """Apply a larger deadzone to a stick input and make digital/binary instead of analog"""
+    if abs(value) < threshold:
+        return 0
+    return int(copysign(1, value))
+
+
 def is_user_in_group(group_name: str) -> bool:
     # Copied from https://zetcode.com/python/os-getgrouplist/
     try:
@@ -344,20 +688,26 @@ def is_user_in_group(group_name: str) -> bool:
         return False
 
 
+def exit_handler(signum, frame):
+    print("Caught SIGTERM. Exiting...")
+    rclpy.try_shutdown()
+    sys.exit(0)
+
+
 def main(args=None):
     try:
         rclpy.init(args=args)
 
+        # Catch termination signals and exit cleanly
+        signal.signal(signal.SIGTERM, exit_handler)
+
         node = Headless()
         rclpy.spin(node)
     except (KeyboardInterrupt, ExternalShutdownException):
         print("Caught shutdown signal. Exiting...")
     finally:
-        rclpy.shutdown()
+        rclpy.try_shutdown()
 
 
 if __name__ == "__main__":
-    signal.signal(
-        signal.SIGTERM, lambda signum, frame: sys.exit(0)
-    )  # Catch termination signals and exit cleanly
     main()

src/latency_tester/src/embedded_ping.cpp

@@ -26,7 +26,7 @@ class LatencyTester : public rclcpp::Node
 {
 public:
     LatencyTester()
-    : Node("latency_tester"), count_(0), target_mcu_("core")
+    : Node("latency_tester"), count_(0)
     {
         publisher_ = this->create_publisher<std_msgs::msg::String>("/anchor/relay", 10);
         timer_ = this->create_wall_timer(
@@ -35,6 +35,8 @@ public:
             "/anchor/debug",
             10,
             std::bind(&LatencyTester::response_callback, this, std::placeholders::_1));
+
+        target_mcu_ = this->declare_parameter<std::string>("target_mcu", "core");
     }
 
 private:

src/servo_arm_twist_pkg/CMakeLists.txt

@@ -1,80 +0,0 @@
-cmake_minimum_required(VERSION 3.22)
-project(servo_arm_twist_pkg)
-
-# C++ Libraries #################################################
-
-# Core C++ library for calculations and collision checking.
-# Provides interface used by the component node.
-set(SERVO_LIB_NAME servo_arm_twist_lib)
-
-# Pose Tracking
-set(POSE_TRACKING pose_tracking)
-
-# Component Nodes (Shared libraries) ############################
-set(SERVO_COMPONENT_NODE servo_node)
-set(SERVO_CONTROLLER_INPUT servo_controller_input)
-
-# Executable Nodes ##############################################
-set(SERVO_NODE_MAIN_NAME servo_node_main)
-set(POSE_TRACKING_DEMO_NAME servo_pose_tracking_demo)
-set(FAKE_SERVO_CMDS_NAME fake_command_publisher)
-
-#################################################################
-
-# Common cmake code applied to all moveit packages
-find_package(moveit_common REQUIRED)
-moveit_package()
-
-set(THIS_PACKAGE_INCLUDE_DEPENDS
-  control_msgs
-  control_toolbox
-  geometry_msgs
-  moveit_core
-  moveit_msgs
-  moveit_ros_planning
-  pluginlib
-  rclcpp
-  rclcpp_components
-  sensor_msgs
-  std_msgs
-  std_srvs
-  tf2_eigen
-  trajectory_msgs
-)
-
-find_package(ament_cmake REQUIRED)
-find_package(eigen3_cmake_module REQUIRED)
-find_package(Eigen3 REQUIRED)
-foreach(Dependency IN ITEMS ${THIS_PACKAGE_INCLUDE_DEPENDS})
-  find_package(${Dependency} REQUIRED)
-endforeach()
-
-#####################
-## Component Nodes ##
-#####################
-
-# Add executable for using a controller
-add_library(${SERVO_CONTROLLER_INPUT} SHARED src/joystick_twist.cpp)
-ament_target_dependencies(${SERVO_CONTROLLER_INPUT} ${THIS_PACKAGE_INCLUDE_DEPENDS})
-rclcpp_components_register_nodes(${SERVO_CONTROLLER_INPUT} "servo_arm_twist_pkg::JoyToServoPub")
-
-#############
-## Install ##
-#############
-
-# Install Libraries
-install(
-  TARGETS
-    ${SERVO_CONTROLLER_INPUT}
-  EXPORT export_${PROJECT_NAME}
-  LIBRARY DESTINATION lib
-  ARCHIVE DESTINATION lib
-  RUNTIME DESTINATION bin
-  INCLUDES DESTINATION include
-)
-
-# Install Binaries
-ament_export_targets(export_${PROJECT_NAME} HAS_LIBRARY_TARGET)
-ament_export_dependencies(${THIS_PACKAGE_INCLUDE_DEPENDS})
-
-ament_package()

src/servo_arm_twist_pkg/README.md

@@ -1,3 +0,0 @@
-# Moveit Servo
-
-See the [Realtime Arm Servoing Tutorial](https://moveit.picknik.ai/main/doc/realtime_servo/realtime_servo_tutorial.html) for installation instructions, quick-start guide, an overview about `moveit_servo`, and to learn how to set it up on your robot.

src/servo_arm_twist_pkg/package.xml

@@ -1,58 +0,0 @@
-<?xml version="1.0"?>
-<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
-<package format="3">
-  <name>servo_arm_twist_pkg</name>
-  <version>2.5.9</version>
-  <description>Provides real-time manipulator Cartesian and joint servoing.</description>
-  <maintainer email="blakeanderson@utexas.edu">Blake Anderson</maintainer>
-  <maintainer email="andyz@utexas.edu">Andy Zelenak</maintainer>
-  <maintainer email="tyler@picknik.ai">Tyler Weaver</maintainer>
-  <maintainer email="henningkayser@picknik.ai">Henning Kayser</maintainer>
-
-  <license>BSD 3-Clause</license>
-
-  <url type="website">https://ros-planning.github.io/moveit_tutorials</url>
-
-  <author>Brian O'Neil</author>
-  <author email="andyz@utexas.edu">Andy Zelenak</author>
-  <author>Blake Anderson</author>
-  <author email="alex@machinekoder.com">Alexander Rössler</author>
-  <author email="tyler@picknik.ai">Tyler Weaver</author>
-  <author email="adam.pettinger@utexas.edu">Adam Pettinger</author>
-
-  <buildtool_depend>ament_cmake</buildtool_depend>
-  <depend>moveit_common</depend>
-
-  <depend>control_msgs</depend>
-  <depend>control_toolbox</depend>
-  <depend>geometry_msgs</depend>
-  <depend>moveit_msgs</depend>
-  <depend>moveit_core</depend>
-  <depend>moveit_ros_planning_interface</depend>
-  <depend>pluginlib</depend>
-  <depend>sensor_msgs</depend>
-  <depend>std_msgs</depend>
-  <depend>std_srvs</depend>
-  <depend>tf2_eigen</depend>
-  <depend>trajectory_msgs</depend>
-
-  <exec_depend>gripper_controllers</exec_depend>
-  <exec_depend>joint_state_broadcaster</exec_depend>
-  <exec_depend>joint_trajectory_controller</exec_depend>
-  <exec_depend>joy</exec_depend>
-  <exec_depend>robot_state_publisher</exec_depend>
-  <exec_depend>tf2_ros</exec_depend>
-  <exec_depend>moveit_configs_utils</exec_depend>
-  <exec_depend>launch_param_builder</exec_depend>
-
-  <test_depend>ament_cmake_gtest</test_depend>
-  <test_depend>ament_lint_auto</test_depend>
-  <test_depend>ament_lint_common</test_depend>
-  <test_depend>controller_manager</test_depend>
-  <test_depend>ros_testing</test_depend>
-
-  <export>
-    <build_type>ament_cmake</build_type>
-  </export>
-
-</package>

src/servo_arm_twist_pkg/src/joystick_twist.cpp

@@ -1,271 +0,0 @@
-/*********************************************************************
- * Software License Agreement (BSD License)
- *
- *  Copyright (c) 2020, PickNik Inc.
- *  All rights reserved.
- *
- *  Redistribution and use in source and binary forms, with or without
- *  modification, are permitted provided that the following conditions
- *  are met:
- *
- *   * Redistributions of source code must retain the above copyright
- *     notice, this list of conditions and the following disclaimer.
- *   * Redistributions in binary form must reproduce the above
- *     copyright notice, this list of conditions and the following
- *     disclaimer in the documentation and/or other materials provided
- *     with the distribution.
- *   * Neither the name of PickNik Inc. nor the names of its
- *     contributors may be used to endorse or promote products derived
- *     from this software without specific prior written permission.
- *
- *  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- *  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- *  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
- *  FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
- *  COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
- *  INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
- *  BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
- *  LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
- *  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
- *  LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
- *  ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
- *  POSSIBILITY OF SUCH DAMAGE.
- *********************************************************************/
-
-/*      Title     : joystick_servo_example.cpp
- *      Project   : servo_arm_twist_pkg
- *      Created   : 08/07/2020
- *      Author    : Adam Pettinger
- */
-
-#include <sensor_msgs/msg/joy.hpp>
-#include <geometry_msgs/msg/twist_stamped.hpp>
-#include <control_msgs/msg/joint_jog.hpp>
-#include <std_srvs/srv/trigger.hpp>
-#include <moveit_msgs/msg/planning_scene.hpp>
-#include <rclcpp/client.hpp>
-#include <rclcpp/experimental/buffers/intra_process_buffer.hpp>
-#include <rclcpp/node.hpp>
-#include <rclcpp/publisher.hpp>
-#include <rclcpp/qos.hpp>
-#include <rclcpp/qos_event.hpp>
-#include <rclcpp/subscription.hpp>
-#include <rclcpp/time.hpp>
-#include <rclcpp/utilities.hpp>
-#include <thread>
-
-// We'll just set up parameters here
-const std::string JOY_TOPIC = "/joy";
-const std::string TWIST_TOPIC = "/servo_node/delta_twist_cmds";
-const std::string JOINT_TOPIC = "/servo_node/delta_joint_cmds";
-const std::string EEF_FRAME_ID = "End_Effector";
-const std::string BASE_FRAME_ID = "base_link";
-
-// Enums for button names -> axis/button array index
-// For XBOX 1 controller
-enum Axis
-{
-    LEFT_STICK_X = 0,
-    LEFT_STICK_Y = 1,
-    LEFT_TRIGGER = 2,
-    RIGHT_STICK_X = 3,
-    RIGHT_STICK_Y = 4,
-    RIGHT_TRIGGER = 5,
-    D_PAD_X = 6,
-    D_PAD_Y = 7
-};
-enum Button
-{
-    A = 0,
-    B = 1,
-    X = 2,
-    Y = 3,
-    LEFT_BUMPER = 4,
-    RIGHT_BUMPER = 5,
-    CHANGE_VIEW = 6,
-    MENU = 7,
-    HOME = 8,
-    LEFT_STICK_CLICK = 9,
-    RIGHT_STICK_CLICK = 10
-};
-
-// Some axes have offsets (e.g. the default trigger position is 1.0 not 0)
-// This will map the default values for the axes
-std::map<Axis, double> AXIS_DEFAULTS = { { LEFT_TRIGGER, 1.0 }, { RIGHT_TRIGGER, 1.0 } };
-std::map<Button, double> BUTTON_DEFAULTS;
-
-// To change controls or setup a new controller, all you should to do is change the above enums and the follow 2
-// functions
-/** \brief // This converts a joystick axes and buttons array to a TwistStamped or JointJog message
- * @param axes The vector of continuous controller joystick axes
- * @param buttons The vector of discrete controller button values
- * @param twist A TwistStamped message to update in prep for publishing
- * @param joint A JointJog message to update in prep for publishing
- * @return return true if you want to publish a Twist, false if you want to publish a JointJog
- */
-bool convertJoyToCmd(const std::vector<float>& axes, const std::vector<int>& buttons,
-                     std::unique_ptr<geometry_msgs::msg::TwistStamped>& twist,
-                     std::unique_ptr<control_msgs::msg::JointJog>& joint)
-{
-    // // Give joint jogging priority because it is only buttons
-    // // If any joint jog command is requested, we are only publishing joint commands
-    // if (buttons[A] || buttons[B] || buttons[X] || buttons[Y] || axes[D_PAD_X] || axes[D_PAD_Y])
-    // {
-    //   // Map the D_PAD to the proximal joints
-    //   joint->joint_names.push_back("panda_joint1");
-    //   joint->velocities.push_back(axes[D_PAD_X]);
-    //   joint->joint_names.push_back("panda_joint2");
-    //   joint->velocities.push_back(axes[D_PAD_Y]);
-
-    //   // Map the diamond to the distal joints
-    //   joint->joint_names.push_back("panda_joint7");
-    //   joint->velocities.push_back(buttons[B] - buttons[X]);
-    //   joint->joint_names.push_back("panda_joint6");
-    //   joint->velocities.push_back(buttons[Y] - buttons[A]);
-    //   return false;
-    // }
-
-    // The bread and butter: map buttons to twist commands
-    twist->twist.linear.z = axes[RIGHT_STICK_Y];
-    twist->twist.linear.y = axes[RIGHT_STICK_X];
-
-    double lin_x_right = -0.5 * (axes[RIGHT_TRIGGER] - AXIS_DEFAULTS.at(RIGHT_TRIGGER));
-    double lin_x_left = 0.5 * (axes[LEFT_TRIGGER] - AXIS_DEFAULTS.at(LEFT_TRIGGER));
-    twist->twist.linear.x = lin_x_right + lin_x_left;
-
-    twist->twist.angular.y = axes[LEFT_STICK_Y];
-    twist->twist.angular.x = axes[LEFT_STICK_X];
-
-    double roll_positive = buttons[RIGHT_BUMPER];
-    double roll_negative = -1 * (buttons[LEFT_BUMPER]);
-    twist->twist.angular.z = roll_positive + roll_negative;
-
-    return true;
-}
-
-/** \brief // This should update the frame_to_publish_ as needed for changing command frame via controller
- * @param frame_name Set the command frame to this
- * @param buttons The vector of discrete controller button values
- */
-void updateCmdFrame(std::string& frame_name, const std::vector<int>& buttons)
-{
-    if (buttons[CHANGE_VIEW] && frame_name == EEF_FRAME_ID)
-        frame_name = BASE_FRAME_ID;
-    else if (buttons[MENU] && frame_name == BASE_FRAME_ID)
-        frame_name = EEF_FRAME_ID;
-}
-
-namespace servo_arm_twist_pkg
-{
-class JoyToServoPub : public rclcpp::Node
-{
-public:
-    JoyToServoPub(const rclcpp::NodeOptions& options)
-        : Node("joy_to_twist_publisher", options), frame_to_publish_(BASE_FRAME_ID)
-    {
-        // Setup pub/sub
-        joy_sub_ = this->create_subscription<sensor_msgs::msg::Joy>(
-            JOY_TOPIC, rclcpp::SystemDefaultsQoS(),
-            [this](const sensor_msgs::msg::Joy::ConstSharedPtr& msg) { return joyCB(msg); });
-
-        twist_pub_ = this->create_publisher<geometry_msgs::msg::TwistStamped>(TWIST_TOPIC, rclcpp::SystemDefaultsQoS());
-        joint_pub_ = this->create_publisher<control_msgs::msg::JointJog>(JOINT_TOPIC, rclcpp::SystemDefaultsQoS());
-        // collision_pub_ =
-        //     this->create_publisher<moveit_msgs::msg::PlanningScene>("/planning_scene", rclcpp::SystemDefaultsQoS());
-
-        // Create a service client to start the ServoNode
-        servo_start_client_ = this->create_client<std_srvs::srv::Trigger>("/servo_node/start_servo");
-        servo_start_client_->wait_for_service(std::chrono::seconds(1));
-        servo_start_client_->async_send_request(std::make_shared<std_srvs::srv::Trigger::Request>());
-
-        // // Load the collision scene asynchronously
-        // collision_pub_thread_ = std::thread([this]() {
-        //   rclcpp::sleep_for(std::chrono::seconds(3));
-        //   // Create collision object, in the way of servoing
-        //   moveit_msgs::msg::CollisionObject collision_object;
-        //   collision_object.header.frame_id = "panda_link0";
-        //   collision_object.id = "box";
-
-        //   shape_msgs::msg::SolidPrimitive table_1;
-        //   table_1.type = table_1.BOX;
-        //   table_1.dimensions = { 0.4, 0.6, 0.03 };
-
-        //   geometry_msgs::msg::Pose table_1_pose;
-        //   table_1_pose.position.x = 0.6;
-        //   table_1_pose.position.y = 0.0;
-        //   table_1_pose.position.z = 0.4;
-
-        //   shape_msgs::msg::SolidPrimitive table_2;
-        //   table_2.type = table_2.BOX;
-        //   table_2.dimensions = { 0.6, 0.4, 0.03 };
-
-        //   geometry_msgs::msg::Pose table_2_pose;
-        //   table_2_pose.position.x = 0.0;
-        //   table_2_pose.position.y = 0.5;
-        //   table_2_pose.position.z = 0.25;
-
-        //   collision_object.primitives.push_back(table_1);
-        //   collision_object.primitive_poses.push_back(table_1_pose);
-        //   collision_object.primitives.push_back(table_2);
-        //   collision_object.primitive_poses.push_back(table_2_pose);
-        //   collision_object.operation = collision_object.ADD;
-
-        //   moveit_msgs::msg::PlanningSceneWorld psw;
-        //   psw.collision_objects.push_back(collision_object);
-
-        //   auto ps = std::make_unique<moveit_msgs::msg::PlanningScene>();
-        //   ps->world = psw;
-        //   ps->is_diff = true;
-        //   collision_pub_->publish(std::move(ps));
-        // });
-    }
-
-    // ~JoyToServoPub() override
-    // {
-    //     if (collision_pub_thread_.joinable())
-    //         collision_pub_thread_.join();
-    // }
-
-    void joyCB(const sensor_msgs::msg::Joy::ConstSharedPtr& msg)
-    {
-        // Create the messages we might publish
-        auto twist_msg = std::make_unique<geometry_msgs::msg::TwistStamped>();
-        auto joint_msg = std::make_unique<control_msgs::msg::JointJog>();
-
-        // This call updates the frame for twist commands
-        updateCmdFrame(frame_to_publish_, msg->buttons);
-
-        // Convert the joystick message to Twist or JointJog and publish
-        if (convertJoyToCmd(msg->axes, msg->buttons, twist_msg, joint_msg))
-        {
-            // publish the TwistStamped
-            twist_msg->header.frame_id = frame_to_publish_;
-            twist_msg->header.stamp = this->now();
-            twist_pub_->publish(std::move(twist_msg));
-        }
-        // else
-        // {
-        //     // publish the JointJog
-        //     joint_msg->header.stamp = this->now();
-        //     joint_msg->header.frame_id = "panda_link3";
-        //     joint_pub_->publish(std::move(joint_msg));
-        // }
-    }
-
-private:
-    rclcpp::Subscription<sensor_msgs::msg::Joy>::SharedPtr joy_sub_;
-    rclcpp::Publisher<geometry_msgs::msg::TwistStamped>::SharedPtr twist_pub_;
-    rclcpp::Publisher<control_msgs::msg::JointJog>::SharedPtr joint_pub_;
-    rclcpp::Publisher<moveit_msgs::msg::PlanningScene>::SharedPtr collision_pub_;
-    rclcpp::Client<std_srvs::srv::Trigger>::SharedPtr servo_start_client_;
-
-    std::string frame_to_publish_;
-
-    // std::thread collision_pub_thread_;
-};  // class JoyToServoPub
-
-}  // namespace servo_arm_twist_pkg
-
-// Register the component with class_loader
-#include <rclcpp_components/register_node_macro.hpp>
-RCLCPP_COMPONENTS_REGISTER_NODE(servo_arm_twist_pkg::JoyToServoPub)

treefmt.nix

@@ -0,0 +1,8 @@
+{ pkgs, ... }:
+{
+  projectRootFile = "flake.nix";
+  programs = {
+    nixfmt.enable = true;
+    black.enable = true;
+  };
+}