RunLength圧縮を学ぶ -PackBits-

RunLength -PackBits-

前回，圧縮技術の第一歩としてランレングス圧縮を学んだのですが，そのままでは使い勝手の悪いものなので今回少し応用を学びます．

ランレングスのデメリットとして，データの内容によっては圧縮どころか膨張してしまう可能性があることがありました．これは，データのラン長が１の時にも1という符号を記録してしまうために起こるものでした．解決策は様々なものが提案されていて，簡単なものですと，ラン長が３以上の時にだけ記録するように組んだりするなどがあります．その中でも実際にも使用されている有名なものを今回やっていこうと思います．

PackBits

PackBitsとは，連続したデータと，それ以外の二つに分けて，それぞれでラン長を記録していく方法です．例えば

元データ： AAAAAAABCBCACCCCC
PackBtis : | AAAAAAA | BCBCA | CCCCC |

というように分けて，それぞれのブロックでラン長を記録していくことになります．ラン長の記録の仕方に少し工夫があります．全てのブロックに自然数値で記録してしまうと，どれが連続したデータとそれ以外なのかの判断ができなくなってしまいます．そこで，連続したデータブロックには正の整数，それ以外のブロックには負の整数で記録するようにすれば判断ができます．

圧縮：7A-5BCBCA5C

これで，ばらけているデータが入ってきても，無駄にラン長を記録することなく，連続したデータを圧縮できるというわけです．

しかし，記録する乱調に符号付整数を使用しているために，一度にまとめて記録できる数が半分に減ります．

Switched RunLength Encoding

一度にまとめて記録できる数を元に戻す方法として，Switched RunLength Encodingというものがあります．PackBitsと同じようにデータをまとめるのですが，ラン長を記録する際に，正負で判断するのではなく，まとまったデータと，それ以外のデータが交互にくるものと仮定して記録していく方法です．例えば，まとまったデータが最初にくると決めておいて符号化すると，

7A5BCBCA5C

というように記録でき，デコード側でもまとまったデータが最初にくるというように合わせておくことで，復号できます．問題は，上記の例ではなかったのですが，違う記号が連続でまとまった時の対処です．

AAAABBBBBCDCB

というようなデータがあると，まとまったデータが交互ではないので記録できないように思えてしまうのですが，交互にならなかった場合は間に0を入れて，無理やり交互にします．

4A05B4CDCB

少し処理が複雑になるのですが，問題なく復号できます．

また，インターネットでSwitched RunLength Encodingについて調べていると，PackBitsとは別物として扱われているのですが，私の参考にしている文献では，PackBitsとはデータをまとまったものとばらけているものの二つに区別することであるとのことなので，PackBitsのうちにこの符号化方法が含まれているようになっています．

実装

Pythonを使って実装していきます．バイナリデータを扱うためにstructモジュールをインポートしています． getPackbitsという関数でデータを分ける作業をして，そのデータを使ってそれぞれエンコードしています．

PackBits　get

def getPacBits(data, processPtr):
    run = 0
    code = data[processPtr]
    processPtr += 1
    run += 1

    if code == data[processPtr]:
        en_mode = 0
        while processPtr < len(data) and code == data[processPtr] and run < RUN_LIMIT:
            code = data[processPtr]
            processPtr += 1
            run += 1
    else:
        en_mode = 1
        while processPtr < len(data) and code != data[processPtr] and run < RUN_LIMIT:
            code = data[processPtr]
            processPtr += 1
            run += 1

        if processPtr != len(data):
            processPtr -= 1
            run -= 1
    return (run, processPtr, en_mode)

PackBits encode

def encode_PackBits(data):
    encode_data = b''
    processPtr = 0
    while processPtr < len(data) - 1:
        codePtr = processPtr
        run, processPtr, mode = getPacBits(data, processPtr)
        if mode:
            encode_data += pack('b', -run) + pack('>'+str(run)+'s',data[codePtr:processPtr])
        else:
            encode_data += pack('b', run) + pack('B', data[codePtr])

    return encode_data

PackBits decode

def decode_PackBits(data):
    decode_data = b''
    processPtr = 0

    while processPtr < len(data) - 1:
        run = unpack_from('b', data, processPtr)[0]
        processPtr += 1
        if run < 0:
            run = abs(run)
            decode_data += pack('>' + str(run) + 's', data[processPtr:processPtr+run])
            processPtr += run
        else:
            decode_data += pack('B', data[processPtr]) * run
            processPtr += 1

    return decode_data

Switched RunLengthEncoding encode

def encode_srle(data):
    encode_data = b''
    processPtr = 0
    mode = 0
    while processPtr < len(data) - 1:
        basePtr = processPtr
        run, processPtr, en_mode = getPacBits(data,processPtr)
        if mode == 0:
            mode = 1
            if en_mode == 1:
                encode_data += pack('B', 0)
                processPtr = basePtr
                continue
            encode_data += pack('B', run) + pack('B', data[basePtr])
        else:
            mode = 0
            if en_mode == 0:
                encode_data += pack('B', 0)
                processPtr = basePtr
                continue
            encode_data += pack('B', run) + pack('>'+str(run)+'s',data[basePtr:processPtr])

    return encode_data

Switched RunLengthEncoding decode

def decode_srle(data):
    decode_data = b''
    processPtr = 0
    mode = 0

    while processPtr < len(data) - 1:
        run = unpack_from('B', data, processPtr)[0]
        processPtr += 1
        if run == 0:
            mode ^= 1
            continue
        if mode == 1:
            decode_data += pack('>' + str(run) + 's', data[processPtr:processPtr+run])
            processPtr += run
            mode = 0
        else:
            decode_data += pack('B', data[processPtr]) * run
            processPtr += 1
            mode = 1

    return decode_data