one more attempt at manual zoom

merge main into ptz-zoom

src/anchor_pkg/anchor_pkg/anchor_node.py

@@ -92,7 +92,7 @@ class SerialRelay(Node):
                     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)
-                elif output.startswith("can_relay_fromvic,citadel") or output.startswith("can_relay_fromvic,digit"):  # digit for SHT sensor
+                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

src/arm_pkg/arm_pkg/arm_node.py

@@ -165,7 +165,7 @@ class SerialRelay(Node):
 
         return
     
-    def send_cmd(self, msg):
+    def send_cmd(self, msg: str):
         if self.launch_mode == 'anchor': #if in anchor mode, send to anchor node to relay
             output = String()
             output.data = msg
@@ -174,7 +174,7 @@ class SerialRelay(Node):
             self.get_logger().info(f"[Arm to MCU] {msg}")
             self.ser.write(bytes(msg, "utf8"))
 
-    def anchor_feedback(self, msg):
+    def anchor_feedback(self, msg: String):
         output = msg.data
         if output.startswith("can_relay_fromvic,arm,55"):
             #pass
@@ -205,10 +205,10 @@ class SerialRelay(Node):
         self.socket_pub.publish(self.arm_feedback)
         self.digit_pub.publish(self.digit_feedback)
 
-    def updateAngleFeedback(self, msg: String):
+    def updateAngleFeedback(self, msg: str):
                 # Angle feedbacks,
         #split the msg.data by commas
-        parts = msg.data.split(",")
+        parts = msg.split(",")
         
         if len(parts) >= 7:
             # Extract the angles from the string
@@ -241,9 +241,9 @@ class SerialRelay(Node):
             self.get_logger().info("Invalid angle feedback input format")
 
 
-    def updateBusVoltage(self, msg: String):
+    def updateBusVoltage(self, msg: str):
         # Bus Voltage feedbacks
-        parts = msg.data.split(",")
+        parts = msg.split(",")
         if len(parts) >= 7:
             # Extract the voltage from the string
             voltages_in = parts[3:7]

src/bio_pkg/bio_pkg/bio_node.py

@@ -144,10 +144,11 @@ class SerialRelay(Node):
 
         # LSS (SCYTHE)
         command = "can_relay_tovic,citadel,24," + str(msg.bio_arm) + "\n"
-        self.send_cmd(command)
+        #self.send_cmd(command)
+        
         # Vibration Motor
-        command = "can_relay_tovic,citadel,26," + str(msg.vibration_motor) + "\n"
-        self.send_cmd(command)
+        command += "can_relay_tovic,citadel,26," + str(msg.vibration_motor) + "\n"
+        #self.send_cmd(command)
         
 
         # FAERIE Control Commands 
@@ -156,16 +157,15 @@ class SerialRelay(Node):
         # To be reviewed before use#
 
         # Laser
-        command = "can_relay_tovic,digit,28," + str(msg.laser) + "\n"
-        self.send_cmd(command)
+        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"
-        print(msg.drill)
-        self.send_cmd(command)
+        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"
+        command += "can_relay_tovic,digit,42," + str(msg.drill_arm) + "\n"
         self.send_cmd(command)
 
 
@@ -182,7 +182,7 @@ class SerialRelay(Node):
     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}")
 
         if output.startswith("can_relay_fromvic,citadel,54"):  # bat, 12, 5, Voltage readings * 100
             self.bio_feedback.bat_voltage = float(parts[3]) / 100.0

src/core_pkg/core_pkg/core_headless.py

@@ -20,113 +20,86 @@ os.environ["SDL_VIDEODRIVER"] = "dummy"  # Prevents pygame from trying to open a
 os.environ["SDL_AUDIODRIVER"] = "dummy"  # Force pygame to use a dummy audio driver before pygame.init()
 
 
-max_speed = 75 #Max speed as a duty cycle percentage (1-100)
+max_speed = 90 #Max speed as a duty cycle percentage (1-100)
 
 class Headless(Node):
     def __init__(self):
-        # Initalize node with name
-        super().__init__("core_headless")
-
-        self.create_timer(0.20, self.send_controls)
-
-
-        # Create a publisher to publish any output the pico sends
-        self.publisher = self.create_publisher(CoreControl, '/core/control', 10) 
-
-
-        self.lastMsg = String() #Used to ignore sending controls repeatedly when they do not change
-
+        # Initialize pygame first
         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
+        
+        # Wait for a gamepad to be connected
+        self.gamepad = None
+        print("Waiting for gamepad connection...")
+        while pygame.joystick.get_count() == 0:
+            # Process any pygame events to keep it responsive
+            for event in pygame.event.get():
+                if event.type == pygame.QUIT:
+                    pygame.quit()
+                    sys.exit(0)
+            time.sleep(1.0)  # Check every second
+            print("No gamepad found. Waiting...")
+        
+        # Initialize the gamepad
         self.gamepad = pygame.joystick.Joystick(0)
         self.gamepad.init()
         print(f'Gamepad Found: {self.gamepad.get_name()}')
-        #
-        #
-
+        
+        # Now initialize the ROS2 node
+        super().__init__("core_headless")
+        self.create_timer(0.15, self.send_controls)
+        self.publisher = self.create_publisher(CoreControl, '/core/control', 10)
+        self.lastMsg = String()  # Used to ignore sending controls repeatedly when they do not change
 
     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.send_controls()
-
-                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()
-                    self.gamepad = pygame.joystick.Joystick(0)
-                    self.gamepad.init() #re-initialized gamepad
-                    print(f"Gamepad reconnected: {self.gamepad.get_name()}")
-                    
-
+                time.sleep(0.1)  # Small delay to avoid CPU hogging
         except KeyboardInterrupt:
             sys.exit(0)
-        
 
     def send_controls(self):
-        
         for event in pygame.event.get():
             if event.type == pygame.QUIT:
                 pygame.quit()
-                exit()
-        input = CoreControl()
-
-        input.max_speed = max_speed
-
-        input.right_stick = round(self.gamepad.get_axis(4),2)#right y-axis
-        if self.gamepad.get_axis(5) > 0:
-            input.left_stick = input.right_stick
-        else:
-            input.left_stick = round(self.gamepad.get_axis(1),2)#lext y-axis
-
-
-
-        if pygame.joystick.get_count() != 0:
-            output = f'L: {input.left_stick}, R: {input.right_stick}, M: {max_speed}' #stop the rover if there is no controller connected
-            self.get_logger().info(f"[Ctrl] {output}")
-            self.publisher.publish(input)
-        else:
+                sys.exit(0)
+        
+        # Check if controller is still connected
+        if pygame.joystick.get_count() == 0:
+            print("Gamepad disconnected. Exiting...")
+            # Send one last zero control message
+            input = CoreControl()
             input.left_stick = 0
             input.right_stick = 0
             input.max_speed = 0
-            self.get_logger().info(f"[Ctrl] Stopping. No Gamepad Connected. ")
             self.publisher.publish(input)
+            self.get_logger().info("Final stop command sent. Shutting down.")
+            # Clean up
+            pygame.quit()
+            sys.exit(0)
+            
+        input = CoreControl()
+        input.max_speed = max_speed
+        input.right_stick = -1 * round(self.gamepad.get_axis(4), 2)  # right y-axis
+        if self.gamepad.get_axis(5) > 0:
+            input.left_stick = input.right_stick
+        else:
+            input.left_stick = -1 * round(self.gamepad.get_axis(1), 2)  # left y-axis
 
-
-        
+        output = f'L: {input.left_stick}, R: {input.right_stick}, M: {max_speed}'
+        self.get_logger().info(f"[Ctrl] {output}")
+        self.publisher.publish(input)
 
 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/core_pkg/core_pkg/core_node.py

@@ -202,6 +202,9 @@ class SerialRelay(Node):
             self.core_feedback.gps_long = float(parts[3])
         elif output.startswith("can_relay_fromvic,core,50"):#GNSS Satellite Count
             self.core_feedback.gps_sats = round(float(parts[3]))
+            #if parts length is at least 5 then we should have altitude, this is a temporary check until fully implemented
+            if len(parts) >= 5:
+                self.core_feedback.gps_alt = round(float(parts[4]), 2)
         elif output.startswith("can_relay_fromvic,core,51"):#Gyro x,y,z
             self.core_feedback.bno_gyro.x = float(parts[3])
             self.core_feedback.bno_gyro.y = float(parts[4])

src/core_pkg/core_pkg/core_ptz.py

@@ -42,13 +42,6 @@ class PtzNode(Node):
         self.loop = None
         self.thread_pool = None
         
-        # Track current zoom level and zoom state
-        self.current_zoom_level = 1.0
-        self.max_zoom_level = 6.0  # A8 mini max zoom
-        self.zoom_step = 0.2       # Zoom step per command
-        self.zooming = 0           # Current zoom direction: -1=out, 0=stop, 1=in
-        self.zoom_timer = None
-        
         # Create publishers
         self.feedback_pub = self.create_publisher(PtzFeedback, '/ptz/feedback', 10)
         self.debug_pub = self.create_publisher(String, '/ptz/debug', 10)
@@ -137,6 +130,7 @@ class PtzNode(Node):
                     debug_str += str(data)
                 self.get_logger().debug(debug_str)
 
+
     def check_camera_connection(self):
         """Periodically check camera connection and attempt to reconnect if needed."""
         if not self.camera_connected and not self.shutdown_requested:
@@ -187,10 +181,6 @@ class PtzNode(Node):
             if msg.reset:
                 self.get_logger().info("Attempting to reset camera to center position")
                 await self.camera.send_attitude_angles_command(0.0, 0.0)
-                
-                # Also reset zoom to 1x when resetting camera
-                self.current_zoom_level = 1.0
-                await self.camera.send_absolute_zoom_command(1.0)
                 return
                 
             if msg.control_mode == 0:
@@ -211,21 +201,15 @@ class PtzNode(Node):
                 self.get_logger().debug(f"Attempting single axis: axis={axis.name}, angle={angle}")
                 await self.camera.send_single_axis_attitude_command(angle, axis)
 
-            elif msg.control_mode == 3:
-                # Instead of absolute zoom, interpret zoom_level as zoom direction
-                zoom_direction = int(msg.zoom_level) if abs(msg.zoom_level) >= 0.5 else 0
-                zoom_direction = max(-1, min(1, zoom_direction))  # Restrict to -1, 0, 1
+            elif msg.control_mode == 3: 
+                zoom_direction = 0
+                if msg.zoom_level > 0:
+                    zoom_direction = 1  # Zoom in
+                elif msg.zoom_level < 0:
+                    zoom_direction = -1  # Zoom out
                 
-                if zoom_direction != self.zooming:
-                    self.zooming = zoom_direction
-                    self.get_logger().debug(f"Zoom direction changed to {zoom_direction}")
-                    
-                    if zoom_direction == 0:
-                        # Stop zooming
-                        self.get_logger().debug(f"Stopping zoom at level {self.current_zoom_level:.1f}x")
-                    else:
-                        # Start zooming in the specified direction
-                        await self.perform_zoom(zoom_direction)
+                self.get_logger().debug(f"Attempting manual zoom: direction={zoom_direction}")
+                await self.camera.send_manual_zoom_command(zoom_direction)
                 
             if hasattr(msg, 'stream_type') and hasattr(msg, 'stream_freq'):
                 if msg.stream_type > 0 and msg.stream_freq >= 0: 
@@ -242,52 +226,13 @@ class PtzNode(Node):
         except RuntimeError as e: # Catch SDK's "not connected" errors
             self.get_logger().warning(f"SDK command failed (likely not connected): {e}")
             # self.camera_connected will be updated by health/connection checks
+            # self.feedback_msg.error_msg = f"Command failed: {str(e)}" # Already set by health check
+            # self.feedback_pub.publish(self.feedback_msg)
         except Exception as e:
             self.get_logger().error(f"Error processing control command: {e}")
             self.feedback_msg.error_msg = f"Control error: {str(e)}"
             self.feedback_pub.publish(self.feedback_msg) # Publish for other errors
 
-    async def perform_zoom(self, direction):
-        """Perform a zoom operation in the specified direction."""
-        if not self.camera or not self.camera.is_connected:
-            return
-            
-        if direction == 0:
-            return
-            
-        # Calculate new zoom level
-        new_zoom_level = self.current_zoom_level + (direction * self.zoom_step)
-        
-        # Clamp to valid range
-        new_zoom_level = max(1.0, min(self.max_zoom_level, new_zoom_level))
-        
-        # Skip if no change (already at min/max)
-        if new_zoom_level == self.current_zoom_level:
-            if direction > 0:
-                self.get_logger().debug(f"Already at maximum zoom level ({self.max_zoom_level:.1f}x)")
-            else:
-                self.get_logger().debug(f"Already at minimum zoom level (1.0x)")
-            return
-            
-        # Set the new zoom level
-        self.get_logger().debug(f"Zooming from {self.current_zoom_level:.1f}x to {new_zoom_level:.1f}x")
-        self.current_zoom_level = new_zoom_level
-        
-        try:
-            await self.camera.send_absolute_zoom_command(new_zoom_level)
-            
-            # If still zooming in same direction, schedule another zoom step
-            if self.zooming == direction:
-                # Schedule a new zoom step after a short delay if we're not at the limits
-                if (direction > 0 and new_zoom_level < self.max_zoom_level) or \
-                   (direction < 0 and new_zoom_level > 1.0):
-                    # Use create_timer for a non-blocking delay
-                    await asyncio.sleep(0.2)  # 200ms delay between zoom steps
-                    if self.zooming == direction:  # Check if direction hasn't changed
-                        await self.perform_zoom(direction)
-        except Exception as e:
-            self.get_logger().error(f"Error during zoom operation: {e}")
-
     def publish_debug(self, message_text):
         """Publish debug message."""
         msg = String()
@@ -310,6 +255,7 @@ class PtzNode(Node):
             self.get_logger().warning("Asyncio loop not running, cannot execute coroutine.")
             return None
 
+
     async def shutdown_node_async(self):
         """Perform clean shutdown of camera connection."""
         self.shutdown_requested = True

src/core_pkg/core_pkg/siyi_sdk/siyi.py

@@ -64,7 +64,8 @@ class CommandID(Enum):
     """
 
     ROTATION_CONTROL = 0x07
-    ABSOLUTE_ZOOM = 0x0C
+    MANUAL_ZOOM = 0x05  # Added manual zoom command
+    ABSOLUTE_ZOOM = 0x08
     ATTITUDE_ANGLES = 0x0E
     SINGLE_AXIS_CONTROL = 0x41
     DATA_STREAM_REQUEST = 0x25
@@ -320,6 +321,47 @@ class SiyiGimbalCamera:
             f"Sent absolute zoom command with level {zoom_level:.1f}x (original request: {original_requested_zoom:.1f}x)"
         )
 
+    def _build_manual_zoom_packet(self, zoom_direction: int) -> bytes:
+        """
+        Helper to build a manual zoom command packet.
+        
+        :param zoom_direction: Direction to zoom: 1 for in, -1 for out, 0 for stop.
+        :return: Complete packet bytes ready to send.
+        """
+        data_len = 1
+        packet = self._build_packet_header(CommandID.MANUAL_ZOOM, data_len)
+        
+        # Convert the direction to a valid byte value:
+        # -1 (zoom out) -> 2
+        # 0 (stop zoom) -> 0
+        # 1 (zoom in) -> 1
+        if zoom_direction < 0:
+            direction_byte = 2  # Zoom out
+        elif zoom_direction > 0:
+            direction_byte = 1  # Zoom in
+        else:
+            direction_byte = 0  # Stop zoom
+        
+        logger.debug(f"Manual zoom: input direction={zoom_direction}, mapped to byte={direction_byte}")
+        packet.append(direction_byte)
+        return self._finalize_packet(packet)
+
+    async def send_manual_zoom_command(self, zoom_direction: int) -> None:
+        """
+        Send a manual zoom command (0x05) to the gimbal.
+        
+        :param zoom_direction: 1 to zoom in, -1 to zoom out, 0 to stop zooming.
+        """
+        if not self.is_connected or not self.writer:
+            raise RuntimeError(
+                "Socket is not connected or writer is None, cannot send manual zoom command."
+            )
+        packet = self._build_manual_zoom_packet(zoom_direction)
+        logger.debug(f"Manual zoom packet: {packet.hex()}")
+        self.writer.write(packet)
+        await self.writer.drain()
+        logger.debug(f"Sent manual zoom command with direction {zoom_direction}")
+
     async def _read_packet(self):
         if not self.reader:
             raise RuntimeError("Reader is not initialized.")