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

Question

我想读取字节.sys.stdin在textmode中打开,但它有一个可用于读取字节的缓冲区:sys.stdin.buffer.

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

1. 读取缓冲文本sys.stdin,将该文本解码为字节,然后回顾

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

   由于输入流中的不可编码字节而不可接受.

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

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

   peek实际上是未被记录的,并且为您提供了一堆字节,您必须对性能密切切片.

顺便说一句.这个错误实际上只适用于管道:for ./myscript.py <somefile,sys.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)

如果你忘记了这个问题:如何在没有de /编码任何东西的情况下从stdin获取下一个字节,并且不推进流的光标？

Answer 1

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

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

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在可窥探的流上实例化一个新的,并将其安装TextIOWrapper为sys.stdin.

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