Python: итерации и генераторы, чтобы заменить циклы while true?

Question

Давайте начнем с моего вопроса: можете ли вы написать код лучше, чем приведенный ниже?

FRAME_DELIMITER = b'\x0a\x0b\x0c\x0d'
def get_data():
    f = bytearray();

    # detect frame delimiter
    while True:
        f += read_byte()
        if f[-4:] == FRAME_DELIMITER:
            start = len(f)-2
            break

    # read data until next frame delimiter
    while True:
        f += self._read_byte()
        if f[-4:] == FRAME_DELIMITER:
            return f[start:-2]

В двух словах, этот код читает поток данных и возвращает весь кадр. Каждый кадр ограничен 0x0a 0x0b 0x0c.Функция read_byte читает один байт в потоке данных (возможно, было бы удобно получить буфер из x байтов).

Я заглянул в документацию Python, чтобы попытаться написатьэтот код более питоническим способом (и лучшей производительностью?).

Я пришел к генераторам и итераторам.

Мы могли бы представить себе такой генератор:

def my_generator(self):
        while True:
            yield self._read_byte()

и поэкспериментируйте со списком и списками символов, как этот:

f = b''.join(itertools.takewhile(lambda c: c != b'\x03', self.my_generator()))

Но на самом деле я застрял, потому что мне нужно проверить шаблон разделителя, а не только один символ. Не могли бы вы помочь дать мне правильное направление ... Или, может быть, мой код выше именно то, что мне нужно?!

Спасибо!

Answer 1

Испытание, которое вы собираетесь проводить, нецелесообразно без некоторого состояния, но вы можете скрыть состояние в вашем генераторе!

Вы можете заставить свой генератор читать сам фрейм, предполагая, что разделитель является постоянным значением (или вы передаете требуемый разделитель). collections.deque может позволить ему легко сохранять состояние только для последних четырех символов, поэтому он не просто скрывает большое хранилище данных в состоянии:

def read_until_delimiter(self):
    # Note: If FRAME_DELIMITER is bytes, and _read_byte returns len 1 bytes objects
    # rather than ints, you'll need to tweak the definition of frame_as_deque to make it store bytes
    frame_as_deque = collections.deque(FRAME_DELIMITER)
    window = collections.deque(maxlen=len(FRAME_DELIMITER))

    while window != frame_as_deque:
        byte = self._read_byte()
        yield byte
        window.append(byte)  # Automatically ages off bytes to keep constant length after filling

Теперь ваш вызывающий может просто сделать:

f = bytearray(self.read_until_delimiter())
# Or bytearray().join(self.read_until_delimiter()) if reading bytes objects, not ints
start = len(f) - 2

Примечание: я определил maxlen в терминах длины FRAME_DELIMITER;ваш конец разделителя почти никогда не пройдет, потому что вы вырезали последние четыре байта и сравнили их с константой, содержащей только три байтов.

Answer 2

Я думаю, говоря, что лучший код - это код, который не вырезает последовательность bytes вместо умного generator и использует только один цикл while:

# just to simulate your method
data = b'AA\x0a\x0b\x0cBBqfdqsfqsfqsvcwccvxcvvqsfq\x0a\x0b\x0cqsdfqs'
index = -1
def get_bytes():
    # you used two method
    # return read_byte() if count == 2 else self._read_byte()
    global index
    index += 1
    return data[index:index + 1]


FRAME_DELIMITER = b'\x0a\x0b\x0c'
def get_data():
    def update_last_delimiter(v):
        """ update the delemeter with the last readed element"""
        nonlocal last_four_byte
        if len(last_four_byte) < len(FRAME_DELIMITER):
            last_four_byte += v
        else:
            last_four_byte = last_four_byte[1:] + v

    count = 2
    last_four_byte = b''
    while True:
        # because you have two method to extract bytes
        # replace get_bytes() by (read_byte() if count == 2 else self._read_byte())
        update_last_delimiter(get_bytes())

        # only yields items when the first DELIMITER IS FOUND
        if count < 2:
            yield last_four_byte[1:2]

        if last_four_byte == FRAME_DELIMITER:
            count -= 1
            if not count:
                break
            else:
                # when first DELIMITER is found we should yield the [-2] element
                yield last_four_byte[1:2]


print(b''.join(get_data()))
# b'\x0b\x0cBBqfdqsfqsfqsvcwccvxcvvqsfq\n\x0b'

Ключевым моментом здесь является отслеживание последних DELIMITER байтов