树莓派项目中如何实现不同Python脚本间的变量值共享?
解决方案
1. 修复导入模块直接执行业务逻辑的问题
Python导入模块时会默认执行模块内所有顶层无缩进代码,你需要把main.py中的业务逻辑放到if __name__ == '__main__'判断块中,只有直接运行main.py时才会触发这部分代码,导入时仅加载函数定义。
修改后的main.py代码:
# main.py from multiprocessing import Process,Pipe import time def f(child_conn): msg = "Hello" child_conn.send(msg) child_conn.close() if __name__ == '__main__': # 仅直接运行main.py时执行下方业务逻辑 for i in range(1000000): print(i) time.sleep(0.05)
修改完成后运行second.py就不会再触发main.py中的for循环,可正常导入f函数。
2. 适配独立进程状态查询的方案调整
你当前使用的multiprocessing.Pipe仅支持同一父进程启动的子进程之间通信,无法满足「主程序独立后台运行、后续手动启动辅助脚本查询状态」的需求,两个完全独立的Python进程需要使用跨进程通信方案,以下是两种适合树莓派的轻量实现:
方案一:临时文件存储状态(实现最简单)
主程序运行过程中定时将状态变量写入系统临时目录的文件,辅助脚本直接读取该文件输出状态即可,无需复杂通信逻辑。main.py示例:
# main.py import time import json if __name__ == '__main__': for i in range(1000000): # 定时更新状态文件 status = {"current_progress": i} with open("/tmp/my_project_status", "w") as f: json.dump(status, f) print(i) time.sleep(0.05)
状态查询脚本示例:
# status_check.py import json try: with open("/tmp/my_project_status", "r") as f: status = json.load(f) print(f"当前运行进度:{status['current_progress']}") except FileNotFoundError: print("主程序未在运行")
方案二:本地UDP通信(实时性更高)
主程序后台启动本地UDP监听服务,辅助脚本向指定端口发送查询请求,主程序返回当前状态,响应延迟更低。main.py示例:
# main.py import time import socket import threading # 全局状态变量 current_progress = 0 def udp_status_server(): s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # 绑定本地端口,不对外暴露 s.bind(('127.0.0.1', 56789)) while True: _, addr = s.recvfrom(1024) # 返回当前进度 s.sendto(str(current_progress).encode(), addr) if __name__ == '__main__': # 启动后台UDP服务线程 threading.Thread(target=udp_status_server, daemon=True).start() for i in range(1000000): current_progress = i print(i) time.sleep(0.05)
状态查询脚本示例:
# status_check.py import socket s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.settimeout(2) try: s.sendto(b"get_status", ('127.0.0.1', 56789)) progress, _ = s.recvfrom(1024) print(f"当前运行进度:{progress.decode()}") except socket.timeout: print("主程序未在运行")
内容的提问来源于stack exchange,提问作者drobot
相关产品推荐
相关产品推荐

