Python 3:使用readahead从stdin管道读取字节


问题内容

我想读取字节。sys.stdin以textmode打开,但它具有可用于读取字节的缓冲区:sys.stdin.buffer

我的问题是,当我将数据传输到python中时,如果我想预读,我似乎只有2个选项,否则我得到一个 io.UnsupportedOperation: File or stream is not seekable.

  1. 从中读取缓冲的文本sys.stdin,将其解码为字节,然后向后查找

sys.stdin.read(1).decode(); sys.stdin.seek(-1, io.SEEK_CUR)

由于输入流中的字节数太长而无法接受。

  1. 利用peek来从标准输入的缓冲一些字节,切片,要适当的数量,并祈祷,因为peek并不能保证任何事情:它可能会少于或多于您的请求......

sys.stdin.buffer.peek(1)[:1]

peek确实文档不足,它为您提供了许多字节,您需要对这些字节进行性能密集的切片。

顺便说一句 该错误实际上仅适用于以下情况:for ./myscript.py <somefilesys.stdin.buffer支持查找。但sys.stdin对象始终是相同的层次结构:

$ cat testio.py
#!/usr/bin/env python3
from sys import stdin
print(stdin)
print(stdin.buffer)
print(stdin.buffer.raw)"
$ ./testio.py
<_io.TextIOWrapper name='<stdin>' mode='r' encoding='UTF-8'>
<_io.BufferedReader name='<stdin>'>
<_io.FileIO name='<stdin>' mode='rb'>
$ ./testio.py <somefile
[the same as above]
$ echo hi | ./testio.py
[the same as above]

一些最初的想法(例如将字节流包装到随机访问缓冲区中)失败,并出现与上述相同的错误:BufferedRandom(sys.stdin.buffer).seek(0)io.UnsupportedOperation…

最后,为方便起见,我提出:

Python的io类层次结构

IOBase
├RawIOBase
│└FileIO
├BufferedIOBase  (buffers a RawIOBase)
│├BufferedWriter┐ 
│├BufferedReader│
││        └─────┴BufferedRWPair
│├BufferedRandom (implements seeking)
│└BytesIO        (wraps a bytes)
└TextIOBase
 ├TextIOWrapper  (wraps a BufferedIOBase)
 └TextIO         (wraps a str)

万一您忘记了这个问题:我如何从stdin中获取下一个字节,而又不进行任何解码/编码,并且不前进流的游标?


问题答案:

异常不是来自Python,而是来自操作系统,该操作系统不允许在管道上进行搜索。(如果您从常规管道重定向输出,即使它是标准输入,也可以对其进行查找。)这就是为什么即使类相同,在一种情况下而不是另一种情况下出现错误的原因。

用于预读的经典Python 2解决方案是将流包装在您自己的实现预读的流实现中:

class Peeker(object):
    def __init__(self, fileobj):
        self.fileobj = fileobj
        self.buf = cStringIO.StringIO()

    def _append_to_buf(self, contents):
        oldpos = self.buf.tell()
        self.buf.seek(0, os.SEEK_END)
        self.buf.write(contents)
        self.buf.seek(oldpos)

    def peek(self, size):
        contents = self.fileobj.read(size)
        self._append_to_buf(contents)
        return contents

    def read(self, size=None):
        if size is None:
            return self.buf.read() + self.fileobj.read()
        contents = self.buf.read(size)
        if len(contents) < size:
            contents += self.fileobj.read(size - len(contents))
        return contents

    def readline(self):
        line = self.buf.readline()
        if not line.endswith('\n'):
            line += self.fileobj.readline()
        return line

sys.stdin = Peeker(sys.stdin)

在Python
3中,sys.stdin在窥视未解码流的同时支持完整操作是很复杂的-stdin.buffer如上所示,将其包装,然后在可窥视的流上实例化一个新TextIOWrapper对象,并将其安装TextIOWrappersys.stdin

但是,因为你只需要在偷看sys.stdin.buffer,上面的代码将工作得很好,转换后cStringIO.StringIO,以io.BytesIO'\n'b'\n'