конвертировать io.StringIO в io.BytesIO

Ответ 1

Это может быть в общем полезный инструмент для преобразования потока символов в поток байтов, так что здесь идет:

import io

class EncodeIO(io.BufferedIOBase):
  def __init__(self,s,e='utf-8'):
    self.stream=s               # not raw, since it isn't
    self.encoding=e
    self.buf=b""                # encoded but not yet returned
  def _read(self,s): return self.stream.read(s).encode(self.encoding)
  def read(self,size=-1):
    b=self.buf
    self.buf=b""
    if size is None or size<0: return b+self._read(None)
    ret=[]
    while True:
      n=len(b)
      if size<n:
        b,self.buf=b[:size],b[size:]
        n=size
      ret.append(b)
      size-=n
      if not size: break
      b=self._read(min((size+1024)//2,size))
      if not b: break
    return b"".join(ret)
  read1=read

Очевидно, write может быть определена симметрично для декодирования ввода и отправки его в основной поток, хотя тогда вам придется иметь дело с достаточным количеством байтов только для части символа.

Ответ 2

Ответ @foobarna можно улучшить, унаследовав базовый класс io

import io
sio = io.StringIO('wello horld')


class BytesIOWrapper(io.BufferedReader):
    """Wrap a buffered bytes stream over TextIOBase string stream."""

    def __init__(self, text_io_buffer, encoding=None, errors=None, **kwargs):
        super(BytesIOWrapper, self).__init__(text_io_buffer, **kwargs)
        self.encoding = encoding or text_io_buffer.encoding or 'utf-8'
        self.errors = errors or text_io_buffer.errors or 'strict'

    def _encoding_call(self, method_name, *args, **kwargs):
        raw_method = getattr(self.raw, method_name)
        val = raw_method(*args, **kwargs)
        return val.encode(self.encoding, errors=self.errors)

    def read(self, size=-1):
        return self._encoding_call('read', size)

    def read1(self, size=-1):
        return self._encoding_call('read1', size)

    def peek(self, size=-1):
        return self._encoding_call('peek', size)


bio = BytesIOWrapper(sio)
print(bio.read())  # b'wello horld'

Ответ 3

Как некоторые отмечали, вы должны сделать кодирование/декодирование самостоятельно.

Тем не менее, вы можете добиться этого элегантным способом - реализовать свой собственный TextIOWrapper для string => bytes.

Вот такой пример:

class BytesIOWrapper:
    def __init__(self, string_buffer, encoding='utf-8'):
        self.string_buffer = string_buffer
        self.encoding = encoding

    def __getattr__(self, attr):
        return getattr(self.string_buffer, attr)

    def read(self, size=-1):
        content = self.string_buffer.read(size)
        return content.encode(self.encoding)

    def write(self, b):
        content = b.decode(self.encoding)
        return self.string_buffer.write(content)

Который производит вывод как это:

In [36]: bw = BytesIOWrapper(StringIO("some lengt˙˚hyÔstring in here"))

In [37]: bw.read(15)
Out[37]: b'some lengt\xcb\x99\xcb\x9ahy\xc3\x94'

In [38]: bw.tell()
Out[38]: 15

In [39]: bw.write(b'ME')
Out[39]: 2

In [40]: bw.seek(15)
Out[40]: 15

In [41]: bw.read()
Out[41]: b'MEring in here'

Надеюсь, это прояснит ваши мысли!

Ответ 4

Интересно, что, хотя вопрос может показаться разумным, не так-то просто найти практическую причину, по которой мне нужно преобразовать StringIO в BytesIO. Оба в основном являются буферами, и вам обычно требуется только один из них для дополнительных манипуляций с байтами или текстом.

Я могу ошибаться, но я думаю, что ваш вопрос на самом деле заключается в том, как использовать экземпляр BytesIO когда некоторый код, которому вы хотите передать его, ожидает текстовый файл.

В этом случае это общий вопрос, и решением является модуль кодеков.

Два обычных случая его использования следующие:

Составьте объект файла для чтения

In [16]: import codecs, io

In [17]: bio = io.BytesIO(b'qwe\nasd\n')

In [18]: StreamReader = codecs.getreader('utf-8')  # here you pass the encoding

In [19]: wrapper_file = StreamReader(bio)

In [20]: print(repr(wrapper_file.readline()))
'qwe\n'

In [21]: print(repr(wrapper_file.read()))
'asd\n'

In [26]: bio.seek(0)
Out[26]: 0

In [27]: for line in wrapper_file:
    ...:     print(repr(line))
    ...:
'qwe\n'
'asd\n'

Создать объект File для записи

In [28]: bio = io.BytesIO()

In [29]: StreamWriter = codecs.getwriter('utf-8')  # here you pass the encoding

In [30]: wrapper_file = StreamWriter(bio)

In [31]: print('жаба', 'цап', file=wrapper_file)

In [32]: bio.getvalue()
Out[32]: b'\xd0\xb6\xd0\xb0\xd0\xb1\xd0\xb0 \xd1\x86\xd0\xb0\xd0\xbf\n'

In [33]: repr(bio.getvalue().decode('utf-8'))
Out[33]: "'жаба цап\\n'"

Ответ 5

StringIO наследует TextIOBase, который имеет следующее в документации по Python 3:

TextIOBase.buffer

Базовый двоичный буфер (экземпляр BufferedIOBase), с TextIOBase имеет дело TextIOBase. Это не является частью API TextIOBase и может не существовать в некоторых реализациях.

Однако последнее предложение не очень многообещающе, и на самом деле CPython, похоже, не поддерживает его:

>>> io.TextIOBase()
<io.TextIOBase object at 0x00000277468BC0B8>
>>> io.TextIOBase().buffer
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
AttributeError: 'TextIOBase' object has no attribute 'buffer'

Основная ветвь исходного кода CPython на данный момент, когда я пишу это, похоже, тоже не предоставляет buffer. Поскольку это единственный способ получить доступ к внутреннему двоичному состоянию, я не думаю, что есть лучший подход, чем тот, который вы упомянули.

Ответ 6

bio из вашего примера - _io.BytesIO класса _io.BytesIO. Вы использовали 2 раза функцию read().

Я придумал преобразование bytes и один метод read():

sio = io.StringIO('wello horld')
b = bytes(sio.read(), encoding='utf-8')
print(b)

Но второй вариант должен быть еще быстрее:

sio = io.StringIO('wello horld')
b = sio.read().encode()
print(b)