You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.18 17:55:18