Python：開いているストリームにコンテンツを書き込む

Question

私は、オープンバイナリファイルをパラメータとして受け取り、それをEOFまでブロック読み込みを実行するAPIを使用しています。

既存のファイル（io.open mode 'rb'）を開くのではなく、計算された/作成されたコンテンツを書き込むストリームを渡したいと思います。実際には、出力がある単方向パイプです。開いているファイルと交換可能な入力ストリームを介して配信されます。

私はBufferedRWPairを見ましたが、入出力オブジェクトに同じオブジェクトを使用しないという警告に違反していることがわかりました。

誰かが適切な例やより良い提案があれば、それは大歓迎です！

最初

更新 これ確かに厄介な例のショーを読んだ後これが終了し import io buf = io.BufferedRandom(io.BytesIO()) buf.write("a") buf.write("b") buf.flush() while True: print "reading" a = buf.read(1024) if not a: break print "read: {}".format(a) buf.close() ....として私はここのコメントに基づいてBufferedRandom見てきたが、私は明らかに間違って何かをやっています溶液を、独立したリードを維持し、記述する位置

import io 
buf = io.BufferedRandom(io.BytesIO()) 
read = 0 
wrote = 0 
buf.seek(wrote) 
wrote += buf.write(b"a") 
wrote += buf.write(b"b") 
buf.seek(read) 
data = buf.read(1) 
read += len(data) 
buf.seek(wrote) 
wrote += buf.write(b"c") 
print "read: {}".format(data) 
buf.seek(read) 
data = buf.read(512) 
read += len(data) 
wrote += buf.write(b"d") 

buf.seek(wrote) 
wrote += buf.write(b"efghihjlmnop") 
while data: 
    print "read: {}".format(data) 
    buf.seek(read) 
    data = buf.read(1024) 
    read += len(data) 
buf.close() 

Answer 1

コメント：...現在の読み書き位置を自分で管理することなく、ストリームへの読み書きをインターリーブすることができます。

これはio.BufferedRandomの動作です。例えば
しかし、あなたが自分のclass StreamRW(io.BufferedRandom)にロジックをカプセル化することができ、
：

class StreamRW(io.BufferedRandom): 
    def __init__(self, raw): 
     super().__init__(raw) 
     self.seek(0) 

    def read(self, size=1): 
     super().seek(self.read_offset) 
     data = super().read(size) 
     self.read_offset = self.tell() 
     return data 

    def write(self, data): 
     super().seek(self.write_offset) 
     written = super().write(data) 
     self.write_offset = self.tell() 
     return written 

    def seek(self, offset): 
     super().seek(offset) 
     self.read_offset = self.write_offset = self.tell() 

#Usage: 
buf = StreamRW(io.BytesIO()) 
... 

さらに以下のようなコードが、buf.seek(0)なし！

---

ファイルの巻き戻しには、buf.seek(0)を使用する必要があります。
注：binary prefix b""を使用する必要があります。
これは私のために働いている：

import io 
buf = io.BufferedRandom(io.BytesIO()) 
buf.write(b"a") 
buf.write(b"b") 
buf.seek(0) 

while True: 
    print "reading" 
    a = buf.read(1024) 
    if not a: break 
    print "read: {}".format(a) 
buf.close() 

出力：
読む：b'ab」のPythonでテスト

：3.4.2および2.7.9