6

シーケンスの前のバイトがダウンストリーム バイトの解釈を制御するため、ストリームとして解析したいバイト データがあります。したがって、BytesIO は私が望むもののように見えます。しかし、struct モジュールによって提供される機能も使用したいと考えています。しかし、構造体のインターフェースはストリーミングしていません。2つを結婚させる賢い/慣用的な方法はありますか?

例として、データのチャンクの例を次に示します。

b'\n\x00\x02\x90\x10\x00\n\x00\x02`\x10\x00\n\x00\x02\x80\x10\x00'

最初の 4 バイトを unsigned ビッグ エンディアン int (例) としてプルしたいstruct.unpack(fmt='>I'。次のバイトは 0x10 であるため、もう 1 バイトあるはずであり、それが 0x00 であることがわかります。そして、最初からやり直し、次の 4 (0x0A000290) を読み取り、洗浄、すすぎ、繰り返します。各 4 バイト ID の直後に続くバイトは、さまざまなダウンストリーム読み取り (一部のバイト、一部のショート) をトリガーします。

私は次のようなことができます

stream = b'\n\x00\x02\x90\x10\x00\n\x00\x02`\x10\x00\n\x00\x02\x80\x10\x00'
while stream:
    id = struct.unpack('>I', stream[:4])
    stream = stream[4:]
    ...

しかし、それはエレガントとは言えません。

4

2 に答える 2

9

私が一般的にしていることは次のとおりです。

def unpack(stream, fmt):
    size = struct.calcsize(fmt)
    buf = stream.read(size)
    return struct.unpack(fmt, buf)

例えば：

>>> b = io.BytesIO(b'\n\x00\x02\x90\x10\x00\n\x00\x02`\x10\x00\n\x00\x02\x80\x10\x00')
>>> print(unpack(b, '>I'))
(167772816,)
>>> print(unpack(b, '>I'))
(268438016,)
>>> print(unpack(b, '>I'))
(39849984,)
>>> print(unpack(b, '>I'))
(167772800,)
>>> print(unpack(b, '>H'))
(4096,)

ストリーム全体を消費したかどうかを知りたい場合は、いつでもこれを実行できます。

buf = stream.read(1)
if buf:
    raise ValueError("Stream not consumed")

ただし、既に使用している同じ関数を呼び出す方がおそらく簡単です。

>>> def ensure_finished(stream):
...     try:
...         unpack(stream, 'c')
...     except struct.error:
...         pass
...     else:
...         raise ValueError('Stream not consumed')
>>> ensure_finished(b)

read要求されたバイト数よりも少ない可能性があるストリームを使用している場合は、whileループを使用して、EOF まで、または十分なバイト数を取得するまで読み取りと追加を続ける必要があります。それ以外の場合は、これで十分です。

于 2013-07-08T22:48:04.603 に答える
1

structs バッファ APIを使用します。

buf = b'\n\x00\x02…'
offset = 0
id = struct.unpack_from('>I', buf, offset); offset += 4
⋮
x = struct.unpack_from('…', buf, offset)

各操作の後にオフセットを記述したくない場合は、次のように小さなラッパーを記述できます。

class unpacker(object):
    def __init__(self, buf):
        self._buf = buf
        self._offset = 0
    def __call__(self, fmt):
        result = struct.unpack_from(fmt, self._buf, self._offset)
        self._offset += struct.calcsize(fmt)
        return result

⋮

unpack = unpacker(buf)
id = unpack('>I')
⋮
x = unpack('…')
于 2013-07-08T22:49:14.240 に答える