Как использовать io для генерации в памяти потоков данных в виде файловых объектов? - PullRequest
Новая
       23

Как использовать io для генерации в памяти потоков данных в виде файловых объектов?

0 голосов
/ 05 сентября 2018

Мне нравится генерировать поток данных в памяти (временный файл) в Python. Один поток заполняет поток данными, а другой использует его.

После проверки инструментов io-Core для работы с потоками мне кажется, что модуль io является лучшим выбором для него.

Итак, я привел для меня простой пример: 

#!/usr/local/bin/python3
# encoding: utf-8

import io

if __name__ == '__main__':
    a = io.BytesIO()
    a.write("hello".encode())
    txt = a.read(100)
    txt = txt.decode("utf-8")
    print(txt)

Мой пример не работает. "hello" не записано в и не может быть прочитано после. Так была ли моя ошибка? Как я должен изменить свой код, чтобы получить файл как объект в памяти?

Ответы [ 2 ]

0 голосов
/ 05 сентября 2018

@ dhilmathy и @ShadowRanger отметили, что io.BytesIO() не имеют отдельного указателя чтения и записи.

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

Мое решение до сих пор: 

#!/usr/local/bin/python3
# encoding: utf-8

import io

class memoryStreamIO(io.BytesIO):
    """
    memoryStreamIO

    a in memory file like stream object 
    """

    def __init__(self):
        super().__init__()
        self._wIndex = 0
        self._rIndex = 0
        self._mutex = threading.Lock()

    def write(self, d : bytearray):
        self._mutex.acquire()
        r = super().write(d)
        self._wIndex += len(d)
        self._mutex.release()
        return r

    def read(self, n : int):
        self._mutex.acquire()
        super().seek(self._rIndex)
        r = super().read(n)
        self._rIndex += len(r)
        # now we are checking if we can
        if self._rIndex == self._wIndex:
            super().truncate(0)
            super().seek(0)
            self._rIndex = 0
            self._wIndex = 0
        self._mutex.release()
        return r

    def seek(self, n):
        self._mutex.acquire()
        self._rIndex = n
        r = super().seek(n)
        self._mutex.release()
        return r


if __name__ == '__main__':
    a = streamIO()

    a.write("hello".encode())
    txt = (a.read(100)).decode()
    print(txt)

    a.write("abc".encode())
    txt = (a.read(100)).decode()
    print(txt)
0 голосов
/ 05 сентября 2018

На самом деле это написано; но чтение это проблема. Вы должны ссылаться на класс io.BytesIO . Вы можете получить значение, используя getvalue(). Мол, 

import io

a = io.BytesIO()
a.write("hello".encode())
txt = a.getvalue()
txt = txt.decode("utf-8")
print(txt)
...