ROS中如何将回调函数的变量传递到另一个文件?
问题原因
当你使用from color import last_color时,Python会将color.py中last_color的初始值(空字符串)复制到当前模块的命名空间中,后续color.py内对last_color的修改只会更新原模块里的变量,当前模块的副本不会同步变化,所以你只能拿到初始值。
另外注意原color.py存在缩进错误:if abs(delta_time.total_seconds()) >= 3:块没有缩进在callback函数内,这会导致逻辑完全不执行,必须先修正这个问题。
解决方法
方法1:导入整个模块,通过模块名访问变量
修改另一个文件的代码,直接导入color模块,每次访问last_color时都通过模块名引用,这样就能获取到原模块中最新的值:
import color a = 0 def observe_color(a): if a == 0: return color.last_color # 直接访问原模块的变量,获取最新值 else: return "None"
同时修正color.py的缩进错误:
import time import datetime import rospy from std_msgs.msg import String last_color = "" last_calltime = datetime.strptime("00:00:00","%H:%M:%S") def callback(data): global last_calltime, last_color cube = data.data t = time.localtime() tf = time.strftime("%H:%M:%S",t) current_time = datetime.strptime(tf, "%H:%M:%S") delta_time = current_time - last_calltime # 修正缩进:将判断逻辑放入callback函数内 if abs(delta_time.total_seconds()) >= 3: if "Red" in cube: last_color = "Red" elif "Green" in cube: last_color = "Green" else: last_color = "Blue" t = time.localtime() tf = time.strftime("%H:%M:%S",t) last_calltime = datetime.strptime(tf, "%H:%M:%S") # rospy回调函数不需要返回值,此处return可删除 # return last_color if __name__ == '__main__': try: rospy.init_node('color') rospy.Subscriber ('/topic', String, callback) rospy.spin() except rospy.ROSInterruptException: pass
方法2:用类封装状态(更规范)
将全局变量封装到类中,避免全局变量的副作用,同时保证状态的一致性:
修改color.py:
import time import datetime import rospy from std_msgs.msg import String class ColorTracker: def __init__(self): self.last_color = "" self.last_calltime = datetime.strptime("00:00:00", "%H:%M:%S") def callback(self, data): cube = data.data t = time.localtime() tf = time.strftime("%H:%M:%S", t) current_time = datetime.strptime(tf, "%H:%M:%S") delta_time = current_time - self.last_calltime if abs(delta_time.total_seconds()) >= 3: if "Red" in cube: self.last_color = "Red" elif "Green" in cube: self.last_color = "Green" else: self.last_color = "Blue" t = time.localtime() tf = time.strftime("%H:%M:%S", t) self.last_calltime = datetime.strptime(tf, "%H:%M:%S") # 创建全局实例,供其他模块引用 tracker = ColorTracker() if __name__ == '__main__': try: rospy.init_node('color') rospy.Subscriber('/topic', String, tracker.callback) rospy.spin() except rospy.ROSInterruptException: pass
然后在另一个文件中:
from color import tracker a = 0 def observe_color(a): if a == 0: return tracker.last_color # 通过类实例获取最新颜色值 else: return "None"
方法3:利用ROS服务(符合ROS设计理念)
既然使用了ROS,更推荐通过ROS服务的方式获取最新颜色,避免跨文件变量引用的问题:
修改color.py,添加一个ROS服务:
import time import datetime import rospy from std_msgs.msg import String from std_srvs.srv import Trigger, TriggerResponse class ColorTracker: def __init__(self): self.last_color = "" self.last_calltime = datetime.strptime("00:00:00", "%H:%M:%S") def callback(self, data): cube = data.data t = time.localtime() tf = time.strftime("%H:%M:%S", t) current_time = datetime.strptime(tf, "%H:%M:%S") delta_time = current_time - self.last_calltime if abs(delta_time.total_seconds()) >= 3: if "Red" in cube: self.last_color = "Red" elif "Green" in cube: self.last_color = "Green" else: self.last_color = "Blue" t = time.localtime() tf = time.strftime("%H:%M:%S", t) self.last_calltime = datetime.strptime(tf, "%H:%M:%S") tracker = ColorTracker() def get_color_service(req): # 服务回调,返回当前的last_color return TriggerResponse(success=True, message=tracker.last_color) if __name__ == '__main__': try: rospy.init_node('color') rospy.Subscriber('/topic', String, tracker.callback) # 注册服务 rospy.Service('/get_last_color', Trigger, get_color_service) rospy.spin() except rospy.ROSInterruptException: pass
在另一个文件中调用服务获取颜色:
import rospy from std_srvs.srv import Trigger a = 0 def observe_color(a): if a == 0: rospy.wait_for_service('/get_last_color') try: # 创建服务代理 get_color = rospy.ServiceProxy('/get_last_color', Trigger) resp = get_color() return resp.message except rospy.ServiceException as e: print(f"服务调用失败: {e}") return "None" else: return "None"
内容的提问来源于stack exchange,提问作者charactercapital
相关产品推荐
相关产品推荐

