30

文字列のリストを生成するジェネレーターがあります。Python にファイルのように見えるユーティリティ/アダプタはありますか?

例えば、

>>> def str_fn():
...     for c in 'a', 'b', 'c':
...         yield c * 3
... 
>>> for s in str_fn():
...     print s
... 
aaa
bbb
ccc
>>> stream = some_magic_adaptor(str_fn())
>>> while True:
...    data = stream.read(4)
...    if not data:
...        break
...    print data
aaab
bbcc
c

データは大きく、ストリーム可能である必要があるため (各フラグメントは数キロバイト、ストリーム全体は数十メガバイト)、ストリーム アダプターに渡す前にジェネレーター全体を熱心に評価したくありません。

4

8 に答える 8

23

これを行う「正しい」方法は、標準のPythonio抽象基本クラスから継承することです。ただし、Pythonで生のテキストクラスを提供し、これを任意の種類のバッファ付きリーダーでラップできるようには見えません。

継承するのに最適なクラスはですTextIOBase。これがそのような実装、処理readline、そしてreadパフォーマンスに注意を払っている間です。（要点）

import io

class StringIteratorIO(io.TextIOBase):

    def __init__(self, iter):
        self._iter = iter
        self._left = ''

    def readable(self):
        return True

    def _read1(self, n=None):
        while not self._left:
            try:
                self._left = next(self._iter)
            except StopIteration:
                break
        ret = self._left[:n]
        self._left = self._left[len(ret):]
        return ret

    def read(self, n=None):
        l = []
        if n is None or n < 0:
            while True:
                m = self._read1()
                if not m:
                    break
                l.append(m)
        else:
            while n > 0:
                m = self._read1(n)
                if not m:
                    break
                n -= len(m)
                l.append(m)
        return ''.join(l)

    def readline(self):
        l = []
        while True:
            i = self._left.find('\n')
            if i == -1:
                l.append(self._left)
                try:
                    self._left = next(self._iter)
                except StopIteration:
                    self._left = ''
                    break
            else:
                l.append(self._left[:i+1])
                self._left = self._left[i+1:]
                break
        return ''.join(l)
于 2012-09-26T14:46:05.820 に答える
14

イテレータからチャンクで読み取る必要があるソリューションを次に示します。

class some_magic_adaptor:
  def __init__( self, it ):
    self.it = it
    self.next_chunk = ""
  def growChunk( self ):
    self.next_chunk = self.next_chunk + self.it.next()
  def read( self, n ):
    if self.next_chunk == None:
      return None
    try:
      while len(self.next_chunk)<n:
        self.growChunk()
      rv = self.next_chunk[:n]
      self.next_chunk = self.next_chunk[n:]
      return rv
    except StopIteration:
      rv = self.next_chunk
      self.next_chunk = None
      return rv


def str_fn():
  for c in 'a', 'b', 'c':
    yield c * 3

ff = some_magic_adaptor( str_fn() )

while True:
  data = ff.read(4)
  if not data:
    break
  print data
于 2012-09-26T02:35:10.933 に答える
5

StringIO の問題は、事前にすべてをバッファにロードする必要があることです。ジェネレーターが無限の場合、これは問題になる可能性があります:)

from itertools import chain, islice
class some_magic_adaptor(object):
    def __init__(self, src):
        self.src = chain.from_iterable(src)
    def read(self, n):
        return "".join(islice(self.src, None, n))
于 2012-09-26T02:16:18.240 に答える
2

Matt の回答を見ると、必ずしもすべての読み取りメソッドを実装する必要がないことがわかります。read1次のように説明されています。

基になる生ストリームの read() への最大 1 回の呼び出しで、最大でsizeバイトを読み取って返します...

io.TextIOWrapper次に、たとえば の実装を持つものでラップできますreadline。boto.s3.key.Key例として、読み取り用のイテレーターを実装するS3 (Amazon Simple Storage Service) からの CSV ファイルのストリーミングを次に示します。

import io
import csv

from boto import s3


class StringIteratorIO(io.TextIOBase):

    def __init__(self, iter):
        self._iterator = iter
        self._buffer = ''

    def readable(self):
        return True

    def read1(self, n=None):
        while not self._buffer:
            try:
                self._buffer = next(self._iterator)
            except StopIteration:
                break
        result = self._buffer[:n]
        self._buffer = self._buffer[len(result):]
        return result


conn = s3.connect_to_region('some_aws_region')
bucket = conn.get_bucket('some_bucket')
key = bucket.get_key('some.csv')    

fp = io.TextIOWrapper(StringIteratorIO(key))
reader = csv.DictReader(fp, delimiter = ';')
for row in reader:
    print(row)

アップデート

これは、少し良く見える関連する質問への回答です。を継承io.RawIOBaseしてオーバーライドしますreadinto。Python 3 では十分なので、1 でラップする代わりに でラップIterStreamできio.BufferedReaderますio.TextIOWrapper。Python では 2read1が必要ですが、単純に表現できますreadinto。

于 2017-04-14T08:34:42.680 に答える
-1

これはまさにstringIOの目的です..

>>> import StringIO
>>> some_var = StringIO.StringIO("Hello World!")
>>> some_var.read(4)
'Hell'
>>> some_var.read(4)
'o Wo'
>>> some_var.read(4)
'rld!'
>>>

または、あなたがそれがどのように聞こえるかをしたい場合

Class MyString(StringIO.StringIO):
     def __init__(self,*args):
         StringIO.StringIO.__init__(self,"".join(args))

その後、簡単にできます

xx = MyString(*list_of_strings)
于 2012-09-26T02:08:05.603 に答える