python 3: чтение байтов из stdin pipe с readahead

Исключение исходит не от Python, а от операционной системы, что не позволяет искать в трубах. (Если вы перенаправляете вывод из обычного канала, его можно искать, даже если это стандартный ввод.) Вот почему вы получаете ошибку в одном случае, а не в другом, хотя классы одинаковы.

Классическим решением Python 2 для readahead было бы объединение потока в вашу собственную реализацию потока, которая реализует 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 над вашим peekable потоком и установить этот TextIOWrapper как sys.stdin .

Однако, поскольку вам нужно только заглянуть в sys.stdin.buffer , приведенный выше код будет работать очень хорошо, после изменения cStringIO.StringIO на io.BytesIO и '\n' на b'\n' .

Решение user4815162342, в то время как чрезвычайно полезно, похоже, имеет проблему в том, что оно отличается от текущего поведения метода peek io.BufferedReader.

Встроенный метод возвращает те же данные (начиная с текущей позиции чтения) для последовательных вызовов peek ().

Решение user4815162342 вернет последовательные фрагменты данных для каждого последовательного вызова. Это подразумевает, что пользователь должен снова заглянуть, чтобы объединить вывод, если они хотят использовать одни и те же данные более одного раза.

Вот исправление для возврата встроенного поведения:

 def _buffered(self): oldpos = self.buf.tell() data = self.buf.read() self.buf.seek(oldpos) return data def peek(self, size): buf = self._buffered()[:size] if len(buf) < size: contents = self.fileobj.read(size - len(buf)) self._append_to_buf(contents) return self._buffered() return buf 

Смотрите полную версию здесь

Существуют и другие оптимизации, которые могут быть применены, например, удаление ранее буферизованных данных при вызове чтения, который исчерпывает буфер. Текущая реализация оставляет любые занесенные данные в буфер, но эти данные недоступны.