「XOR をとるコーデック」のファクトリ

codecs モジュールには先日までは知らなかった、 IncrementalDecoder により一字ずつデコードする機能があった。これにはまだまだ知らないで損している機能があるような気がして、もうすこしさわってみることにした。

まずはコーデックの自作と登録から。それもなるべく簡単なヤツで。となると換字式暗号、それも XOR あたりがいいかな。これなら変換部分の実装で困ることはないため、どのような動作をする callable オブジェクトを用意すべきか、そのためにどのようなものが用意されているのか調べるほうに集中できるはず。

そうしてできたのが以下の XOR コーデックを作って登録するコード。 factory_xor_codec(XORする値, コーデックの名前) でコーデックが作成される。register_xor_codec(XORする値, コーデックの名前) は作成と登録を同時に行う。たとえば register_xor_codec(0xff, 'allinvert') なら全ビットを反転する 'allinvert' コーデックが作成、登録される。

以下、手順メモ。

まずは encode, decode メソッドを持つ自作 Codecs クラスを作成。その次に StreamWriter, StreamReader を作るのだが自力で全部実装すると、しんどそう。そのためか汎用的な動作を定義済みの codecs.StreamWriter, codecs.StreamReader が用意されていることがわかった。自作コーデックとこれを多重継承するだけでまずは OK 。余力があれば IncrementalEncoder, IncrementalDecoder も作る。

これらを引数に codecs.CodecInfo オブジェクトを作成する。

最後にコーデックを使用できるよう、登録する。名前を受け取って対応する CodecInfo オブジェクトを返す関数を作成し、 codecs.register を実行すればよい。

以下、課題。
codecs.StreamWriter, codecs.StreamReader のコードを読む。

# coding: utf-8
u""" 「XOR をとるコーデック」のファクトリ
"""

import codecs

def factory_xor_codec(bits, name_):
    if not 0x00 <= bits <= 0xff:
        raise ValueError('bits not in range(256)')
    class Codec(codecs.Codec):
        def encode(self, input, errors='strict'):
            output = ''.join((chr(ord(c) ^ bits)) for c in input)
            return output, len(output)
        decode = encode

    class StreamWriter(Codec, codecs.StreamWriter):
        pass

    class StreamReader(Codec, codecs.StreamReader):
        pass

    class IncrementalEncoder(codecs.IncrementalEncoder):
        def encode(self, input, final=False):
            for c in input:
                yield chr(ord(c) ^ bits)

    class IncrementalDecoder(codecs.IncrementalDecoder):
        def decode(self, input, final=False):
            for c in input:
                yield chr(ord(c) ^ bits)

    codec = Codec()
    codec_info = codecs.CodecInfo(
            encode=codec.encode,
            decode=codec.decode,
            streamwriter=StreamWriter,
            streamreader=StreamReader,
            incrementalencoder=IncrementalEncoder,
            incrementaldecoder=IncrementalDecoder,
            name=name_)

    return codec_info

def register_xor_codec(bits, name_):
    CodecInfo = factory_xor_codec(bits, name_)
    def search_codec(name):
        if name == name_:
            return CodecInfo
    codecs.register(search_codec)

def _test():
    register_xor_codec(0xff, 'allinvert')
    bits = ''.join(chr(i) for i in xrange(256))
    print repr(bits.encode('allinvert'))

if __name__ == '__main__':
    _test()