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

树莓派项目中如何实现不同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

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.10.05 13:57:05