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

Я хочу читать байты. sys.stdin открывается в текстовом режиме, но у него есть буфер, который можно использовать для чтения байтов: 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 чтобы получить несколько байтов из буфера stdin, нарезая это на соответствующее число и молясь, поскольку peek ничего не гарантирует: он может дать меньше или больше, чем вы запрашиваете …

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

    peek действительно недооценен и дает вам кучу байтов, которые вы должны выполнить с интенсивным использованием.

Кстати. эта ошибка действительно применима только тогда, когда ./myscript.py <somefile : для ./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…

наконец, для вашего удобства я представляю:

Иерархия классов io класса Python

 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 для 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 

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

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