Przypuśćmy, że mam jakieś StringIO (z cStringIO). Chcę odczytać bufor, dopóki niektóre spotkanie / bajt zostanie napotkany, powiedzmy "Z", więc:

stringio = StringIO('ABCZ123')
buf = read_until(stringio, 'Z')  # buf is now 'ABCZ'
# strinio.tell() is now 4, pointing after 'Z'

Jaki jest najszybszy sposób na to w Pythonie? Dziękuję Ci

7
zaharpopov 26 listopad 2011, 20:25

3 odpowiedzi

Najlepsza odpowiedź

Bardzo rozczarowuję, że to pytanie otrzymuje tylko jedną odpowiedź na przepełnienie stosu, ponieważ jest to interesujące i istotne pytanie. W każdym razie, ponieważ tylko Ovgolovin daje rozwiązanie i myślę, że może to być powoli, myślałem, że szybsze rozwiązanie:

def foo(stringio):
    datalist = []
    while True:
        chunk = stringio.read(256)
        i = chunk.find('Z')
        if i == -1:
            datalist.append(chunk)
        else:
            datalist.append(chunk[:i+1])
            break
        if len(chunk) < 256:
            break
    return ''.join(datalist)

To czyta IO w kawałkach (może char koniec nie znaleziono nie w pierwszym kawałku). Jest bardzo szybki, ponieważ żadna funkcja Pythona wezwała do każdej postaci, ale w przeciwieństwie maksymalne wykorzystanie funkcji Pythona Python.

To przebiegło o 60x szybciej niż roztwór Ovgolovina. I ran timeit, aby go sprawdzić.

7
zaharpopov 27 listopad 2011, 12:17
i = iter(lambda: stringio.read(1),'Z')
buf = ''.join(i) + 'Z'

Tutaj iter jest używany w tym trybie: iter(callable, sentinel) -> iterator.

''.join(...) jest dość skuteczny. Ostatnia operacja dodawania "z '{x1}} nie jest tak dobra. Ale może być adresowany przez dodanie 'Z' do iteratora:

from itertools import chain, repeat

stringio = StringIO.StringIO('ABCZ123')
i = iter(lambda: stringio.read(1),'Z')
i = chain(i,repeat('Z',1))
buf = ''.join(i)

Jeszcze jeden sposób, aby używać generatora:

def take_until_included(stringio):
    while True:
        s = stringio.read(1)
        yield s
        if s=='Z':
            return

i = take_until_included(stringio)
buf = ''.join(i)

Dokonałem pewnych testów wydajności. Wydajność opisanych technik jest ładna ta sama:

http://ideone.com/dqge5.

2
ovgolovin 27 listopad 2011, 11:08
#!/usr/bin/env python3
import io


def iterate_stream(stream, delimiter, max_read_size=1024):
    """ Reads `delimiter` separated strings or bytes from `stream`. """
    empty = '' if isinstance(delimiter, str) else b''
    chunks = []
    while 1:
        d = stream.read(max_read_size)
        if not d:
            break
        while d:
            i = d.find(delimiter)
            if i < 0:
                chunks.append(d)
                break
            chunks.append(d[:i+1])
            d = d[i+1:]
            yield empty.join(chunks)
            chunks = []
    s = empty.join(chunks)
    if s:
        yield s


if __name__ == '__main__':
    print(next(iterate_stream(io.StringIO('ABCZ123'), 'Z')))
    print(next(iterate_stream(io.BytesIO(b'ABCZ123'), b'Z')))
1
pasztorpisti 4 listopad 2019, 17:40