Python 通过subprocess运行代码，并实时获得该代码的窗口输出（如print输出、神经网络模型训练和推理相关实时输出等）

目录

• • • • 1. 需求
      • 2. 不考虑缓存情况下的实现
      • • 2.1. 子进程测试代码
        • 2.2. 主进程代码
      • 3. 考虑缓存情况下的实现

1. 需求

python主进程唤起一个Python子进程，该子进程运行时会实时打印信息，主进程需要获得子进程的实时信息。

这里，需要区分两种情况：当子进程消息较少时，比如自己写的一个子进程，里面用print打印部分信息，此时不需要考虑缓存问题；当子进程消息较多且输出速度快，比如yolov5模型的训练，输出里面存在进度条显示，数据量较大且速度很快，此时需要考虑缓存问题。

主要参考博客《python如何启动新进程,并实时获取程序的输出.》和博客《python subprocess.Popen的使用》

2. 不考虑缓存情况下的实现

此种情况适用于子进程消息较少时，比如自己写的一个子进程，里面用print打印部分信息。

2.1. 子进程测试代码

该代码为子进程测试代码testSub.py，代码内用print进行输出。

import timei = 0while (i<40):    print("当前处于第{}秒".fORMat(i))    time.sleep(1)    i += 1

2.2. 主进程代码

该代码为主进程代码testMain.py，用subprocess.popen运行子进程testSub.py，并获得子进程的实时打印结果。

import subprocessimport threadingclass CMDProcess(threading.Thread):    '''        执行CMD命令行的 进程    '''    def __init__(self, args,callback):        threading.Thread.__init__(self)        self.args = args        self.callback=callback            def run(self):        self.proc = subprocess.Popen(            self.args,            bufsize=0,            shell = False,            stdout=subprocess.PIPE        )                while self.proc.poll() is None:            line = self.proc.stdout.readline()            line = line.decode("utf8")             if(self.callback):                self.callback(line)def getSubInfo(text):    print("子进程测试代码实时输出内容=>" + text)def main():    cmd = [            'python',            '-u', # 注意，这里必须带上-u            'testSub.py'            ]    print("子进程测试代码的运行命令：", ' '.join(cmd))    testProcess = CMDProcess(cmd,getSubInfo )    testProcess.start()main()

3. 考虑缓存情况下的实现

此种情况适用于子进程消息较多且输出速度快，比如yolov5模型的训练，输出里面存在进度条显示，数据量较大且速度很快。

这里，只需要提供主进程代码即可，请注意相关注释内容。该代码为主进程代码testMain.py，用subprocess.popen运行子进程testSub.py，并获得子进程的实时打印结果。

import subprocessimport threadingclass CMDProcess(threading.Thread):    '''        执行CMD命令行的 进程    '''    def __init__(self, args,callback):        threading.Thread.__init__(self)        self.args = args        self.callback=callback        self.cwd = './'             def run(self):        self.proc = subprocess.Popen(            self.args,            bufsize=1, # bufsize=0时，为不缓存；bufsize=1时，按行缓存；bufsize为其他正整数时，为按照近似该正整数的字节数缓存            shell = False,            stdout=subprocess.PIPE,            stderr=subprocess.STDOUT, # 这里可以显示yolov5训练过程中出现的进度条等信息            text=True, # 缓存内容为文本，避免后续编码显示问题            cwd=self.cwd # 这个参数意思是，当前子进程结束后，其结果保存地址，比如yolov5训练进程结束后会输出模型、检测图片等，可在cwd中找到        )                while self.proc.poll() is None:            line = self.proc.stdout.readline()            self.proc.stdout.flush() # 刷新缓存，防止缓存过多造成卡死            #line = line.decode("utf8")             if(self.callback):                self.callback(line)def getSubInfo(text):    print("子进程测试代码实时输出内容=>" + text)def main():    cmd = [            'python',            '-u', # 注意，这里必须带上-u            'testSub.py'            ]      # 这里可以改成yolov5的训练代码，如下    """    cmd = [            'python',            '-u', # 注意，这里必须带上-u            'train.py',            '--img',            '640',            '--batch',            '4',            '--epochs',            '1' #等等，请自行添加参数            ] """    print("子进程测试代码的运行命令：", ' '.join(cmd))    testProcess = CMDProcess(cmd,getSubInfo )    testProcess.start()main()

搭配Qt显示，效果如下

来源地址：https://blog.csdn.net/qq_30841655/article/details/127884956