|
| 1 | +#!/usr/bin/env python3 |
| 2 | +import rclpy |
| 3 | +from rclpy.node import Node |
| 4 | +from pylx16a.lx16a import LX16A |
| 5 | +from ros2_lx16a_driver.srv import GetServoInfo |
| 6 | + |
| 7 | +class LX16AController(Node): |
| 8 | + def __init__(self): |
| 9 | + super().__init__('lx16a_controller') |
| 10 | + |
| 11 | + # Initialisation du bus LX-16A |
| 12 | + try: |
| 13 | + LX16A.initialize("/dev/ttyUSB0") # Remplacez par votre port série |
| 14 | + self.servos = {} |
| 15 | + except Exception as e: |
| 16 | + self.get_logger().error(f"Erreur d'initialisation de la communication : {e}") |
| 17 | + self.destroy_node() |
| 18 | + return |
| 19 | + |
| 20 | + # Création du service GetServoInfo |
| 21 | + self.srv = self.create_service(GetServoInfo, '/get_servo_info', self.handle_get_servo_info) |
| 22 | + |
| 23 | + self.get_logger().info('Nœud LX-16A prêt.') |
| 24 | + |
| 25 | + def get_servo(self, servo_id): |
| 26 | + """Retourne une instance de servo donnée, la crée si nécessaire.""" |
| 27 | + if servo_id not in self.servos: |
| 28 | + try: |
| 29 | + self.servos[servo_id] = LX16A(servo_id) |
| 30 | + self.get_logger().info(f"Servo {servo_id} ajouté.") |
| 31 | + except Exception as e: |
| 32 | + self.get_logger().error(f"Erreur lors de l'ajout du servo {servo_id} : {e}") |
| 33 | + return None |
| 34 | + return self.servos[servo_id] |
| 35 | + |
| 36 | + def handle_get_servo_info(self, request, response): |
| 37 | + """Service pour obtenir les informations détaillées d'un servo.""" |
| 38 | + servo_id = int(request.id) # ID du servo demandé |
| 39 | + servo = self.get_servo(servo_id) |
| 40 | + |
| 41 | + if servo is None: |
| 42 | + self.get_logger().error(f"Servo {servo_id} non disponible.") |
| 43 | + return response # Laisse la réponse vide si le servo n'est pas trouvé |
| 44 | + |
| 45 | + try: |
| 46 | + # Récupération des informations du servo |
| 47 | + response.angle_offset = int(servo.get_angle_offset()) |
| 48 | + response.angle_min = int(servo.get_angle_limits()[0]) |
| 49 | + response.angle_max = int(servo.get_angle_limits()[1]) |
| 50 | + response.voltage_min = float(servo.get_vin_limits()[0]) |
| 51 | + response.voltage_max = float(servo.get_vin_limits()[1]) |
| 52 | + response.temperature_limit = float(servo.get_temp_limit()) |
| 53 | + motor_mode = servo.is_motor_mode(poll_hardware=True) |
| 54 | + response.motor_mode = motor_mode |
| 55 | + if motor_mode: |
| 56 | + response.motor_speed = servo.get_motor_speed() |
| 57 | + response.torque_enabled = servo.is_torque_enabled() |
| 58 | + response.led_enabled = servo.is_led_power_on() |
| 59 | + response.led_error_temp = servo.get_led_error_triggers()[0] |
| 60 | + response.led_error_voltage = servo.get_led_error_triggers()[1] |
| 61 | + response.led_error_locked = servo.get_led_error_triggers()[1] |
| 62 | + response.current_temperature = float(servo.get_temp()) |
| 63 | + response.input_voltage = float(servo.get_vin()) |
| 64 | + if not motor_mode: |
| 65 | + response.physical_angle = servo.get_physical_angle() |
| 66 | + response.commanded_angle = servo.get_commanded_angle() |
| 67 | + |
| 68 | + self.get_logger().info(f"Informations du servo {servo_id} récupérées.") |
| 69 | + except Exception as e: |
| 70 | + self.get_logger().error(f"Erreur lors de la récupération des infos du servo {servo_id} : {e}") |
| 71 | + # Réponse vide si une erreur se produit |
| 72 | + return response |
| 73 | + |
| 74 | + return response |
| 75 | + |
| 76 | + |
| 77 | +def main(args=None): |
| 78 | + rclpy.init(args=args) |
| 79 | + node = LX16AController() |
| 80 | + |
| 81 | + try: |
| 82 | + rclpy.spin(node) |
| 83 | + except KeyboardInterrupt: |
| 84 | + node.get_logger().info('Arrêt du nœud.') |
| 85 | + finally: |
| 86 | + node.destroy_node() |
| 87 | + rclpy.shutdown() |
| 88 | + |
| 89 | +if __name__ == '__main__': |
| 90 | + main() |
0 commit comments