重复写入stdin并从python中读取进程的stdout

Question

我有一段fortran代码从STDIN读取一些数字并将结果写入STDOUT.例如:

do
  read (*,*) x
  y = x*x
  write (*,*) y
enddo

所以我可以从shell启动程序并获得以下输入/输出序列:

5.0
25.0
2.5
6.25

现在我需要在python中执行此操作.与subprocess.Popen徒劳摔跤,并通过本网站的老问题寻找后,我决定用pexpect.spawn:

import pexpect, os
p = pexpect.spawn('squarer')
p.setecho(False)
p.write("2.5" + os.linesep)
res = p.readline()

它的工作原理.问题是,在真实的数据,我需要Python和我的Fortran程序之间传递是10万(或更多)的双精度浮点数的数组.如果它们包含在一个名为的数组中x,那么

p.write(' '.join(["%.10f"%k for k in x]) + os.linesep)

使用pexpect的以下错误消息超时:

buffer (last 100 chars):   
before (last 100 chars):   
after: <class 'pexpect.TIMEOUT'>  
match: None  
match_index: None  
exitstatus: None
flag_eof: False
pid: 8574
child_fd: 3
closed: False
timeout: 30
delimiter: <class 'pexpect.EOF'>
logfile: None
logfile_read: None
logfile_send: None
maxread: 2000
ignorecase: False
searchwindowsize: None
delaybeforesend: 0.05
delayafterclose: 0.1
delayafterterminate: 0.1

除非x少于303个元素.有没有办法将大量数据传入/传出另一个程序的STDIN/STDOUT？

我曾尝试将数据分成更小的块,但后来我的速度损失很大.

提前致谢.

Answer 1

使用子进程模块找到了一个解决方案,所以如果有人需要做同样的事情我在这里发布它以供参考.

import subprocess as sbp

class ExternalProg:

    def __init__(self, arg_list):
        self.opt = sbp.Popen(arg_list, stdin=sbp.PIPE, stdout=sbp.PIPE, shell=True, close_fds=True)

    def toString(self,x):
        return ' '.join(["%.12f"%k for k in x])

    def toFloat(self,x):
        return float64(x.strip().split())

    def sendString(self,string):
        if not string.endswith('\n'):
            string = string + '\n'
        self.opt.stdin.write(string)

    def sendArray(self,x):
        self.opt.stdin.write(self.toString(x)+'\n')

    def readInt(self):
        return int(self.opt.stdout.readline().strip())

    def sendScalar(self,x):
        if type(x) == int:
            self.opt.stdin.write("%i\n"%x)
        elif type(x) == float:
            self.opt.stdin.write("%.12f\n"%x)

    def readArray(self):
        return self.toFloat(self.opt.stdout.readline())

    def close(self):
        self.opt.kill()

使用名为"optimizer"的外部程序调用该类,如下所示:

optim = ExternalProg(['./optimizer'])
optim.sendScalar(500) # send the optimizer the length of the state vector, for example
optim.sendArray(init_x) # the initial guess for x
optim.sendArray(init_g) # the initial gradient g
next_x = optim.readArray() # get the next estimate of x
next_g = evaluateGradient(next_x) # calculate gradient at next_x from within python
# repeat until convergence

在fortran方面(程序编译为提供可执行文件'optimizer'),将读入一个500元素的向量:

read(*,*) input_vector(1:500)

并将写出来:

write(*,'(500f18.11)') output_vector(1:500)

就是这样!我用最多200,000个元素的状态向量测试了它(这是我现在需要的上限).希望这能帮助除我以外的其他人.这个解决方案适用于ifort和xlf90,但由于某些原因我不理解gfortran.

Answer 2

我认为你只需要在这里添加一个换行符：

p.write(' '.join(["%.10f"%k for k in x]) + os.linesep)

而不是每行添加一个。