11.Ros2系列知识之-使用Python编写参数接口

import rclpyfrom rclpy.node import Nodefrom weather_service_demo.srv import GetWeatherimport randomimport timeclass WeatherServer(Node):    def __init__(self):        super().__init__('weather_py_server')        # 1. 定义参数：是否开启随机天气模式        self.declare_parameter('random_mode', True)        # 2. 定义参数：默认城市        self.declare_parameter('default_city', 'Beijing')        self.srv = self.create_service(            GetWeather,            'get_weather',            self.get_weather_callback)        self.get_logger().info(f'Server started. Random mode: True, City: Beijing')    def get_weather_callback(self, request, response):        # 3. 从参数中获取随机天气模式和默认城市        self.random_mode = self.get_parameter('random_mode').value        self.default_city = self.get_parameter('default_city').value        # 线程停1秒，模拟服务端处理时间        time.sleep(1.0)        # 使用参数        if self.random_mode:            weather_options = ['Sunny', 'Rainy', 'Cloudy', 'Snowy']            response.weather_status = random.choice(weather_options)        else:            response.weather_status = 'Fixed_Sunny'        self.get_logger().info(f'Request: {request.date} → Response: {response.weather_status}, City: {self.default_city}')        return responsedef main(args=None):    rclpy.init(args=args)    server = WeatherServer()    rclpy.spin(server)    server.destroy_node()    rclpy.shutdown()if __name__ == '__main__':    main()

import rclpyfrom rclpy.node import Node# 1. 导入自定义的服务接口 (注意这里是导入上一章的包)from weather_service_demo.srv import GetWeatherimport datetimeimport timeclass WeatherClient(Node):    def __init__(self):        super().__init__('weather_py_client')        # 3. 声明并创建客户端        self.cli = self.create_client(GetWeather, 'get_weather')        # 等待服务端上线        while not self.cli.wait_for_service(timeout_sec=1.0):            self.get_logger().info('Service not available, waiting again...')        self.get_logger().info('Weather Client is ready.')    # 2. 创建请求结果接收回调函数    def send_request(self):        # 5. 调用客户端发送请求        # 获取当前日期作为请求参数        today = datetime.datetime.now().strftime("%Y-%m-%d")        request = GetWeather.Request()        request.date = today        # self.get_logger().info(f'Sending request for date: {today}')        # 发送请求并等待响应        future = self.cli.call_async(request)        future.add_done_callback(self.response_callback)    # 4. 编写结果接收逻辑    def response_callback(self, future):        try:            response = future.result()            self.get_logger().info(f'Received weather: {response.weather_status}, City: {response.city}')            # 收到响应后，再次发送下一次请求            self.send_request()        except Exception as e:            self.get_logger().error(f'Service call failed: {e}')def main(args=None):    rclpy.init(args=args)    weather_client = WeatherClient()    # 发送第一次请求    weather_client.send_request()    # 进入事件循环，等待响应并自动发送下一次请求    rclpy.spin(weather_client)    weather_client.destroy_node()    rclpy.shutdown()if __name__ == '__main__':    main()

colcon build --packages-select weather_py_serversource install/setup.bashros2 run weather_py_server weather_py_server

ros2 param set /weather_py_server random_mode False

ros2 param set /weather_server default_city Shanghai