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 文档以尝试以更 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

注意：我根据FRAME_DELIMITER的长度定义了maxlen；您的分隔符结尾几乎永远不会通过，因为您切掉了最后 4 个字节，并将它们与仅包含 3 个字节的常量进行比较。

Answer 2

我认为通过说更好的代码是不是slice串联的​​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字节