今回は「桁上がりのないレンジコーダ」と「適応型レンジコーダ」を取り上げます。
前回説明しましたが、レンジコーダ (Range Coder) は 1998 年に Michael Schindler 氏が発表した方法で、「高性能、高速、特許フリー」の方法として注目を集めるようになりました。レンジコーダは原理的には算術符号と同じ方法で、記号の出現確率だけを利用して記号を符号化します。圧縮性能は算術符号に比べるとわずかに劣化しますが、実現方法はとても簡単で実行速度も高速です。
Michael Schindler 氏のレンジコーダは計算の途中で「桁上がり」が発生しますが、ロシアの Dmitry Subbotin 氏が発表した「桁上げのないレンジコーダ」は、その名のごとく桁上がりが発生しません。現在、レンジコーダは主にこの 2 種類の形式が存在するようです。
今回は桁上がりのないレンジコーダを説明します。そして、実際にファイルを圧縮してみましょう。
それではプログラムを作りましょう。最初に、レンジコーダを表すクラスを定義します。
リスト : 桁上がりのないレンジコーダ
from byteio import *
# 定数
ENCODE = "encode"
DECODE = "decode"
MAX_RANGE = 0xffffffff # 修正 (2007/10/06)
MIN_RANGE = 0x10000
MASK = 0xffffffff
MASK1 = 0xff000000
SHIFT = 24
class RangeCoder:
def __init__(self, s, mode):
self.stream = s
self.range_ = MAX_RANGE
self.low = 0
if mode == DECODE:
# 4 byte read
self.code = self.stream.getc()
self.code = (self.code << 8) + self.stream.getc()
self.code = (self.code << 8) + self.stream.getc()
self.code = (self.code << 8) + self.stream.getc()
elif mode != ENCODE:
raise "RangeCoder mode error"
インスタンス変数 range_ と low は前回と同じく幅と下限値を表します。桁上がりは発生しないので、インスタンス変数 buff と cnt は不要になります。インスタンス変数 code は復号のときに使います。code はファイルから 4 バイト読み込んで初期化します。code の説明は復号のプログラムを作るときに行います。
次は出現頻度表を表すクラス Freq を定義します。
リスト : 出現頻度表
class Freq:
def __init__(self, count):
size = len(count)
self.size = size
m = sum(count)
# 正規化
if m >= MIN_RANGE:
self.count = [0] * size
n = 0
while m >= MIN_RANGE:
m >>= 1
n += 1
for x in range(size):
if count[x] != 0:
self.count[x] = (count[x] >> n) | 1
else:
self.count = count[:]
self.count_sum = [0] * (size + 1)
# 累積度数表
for x in range(size):
self.count_sum[x + 1] = self.count_sum[x] + self.count[x]
桁上がりのないレンジコーダの場合、MIN_RANGE を大きな値にすると圧縮率が劣化することがあります。実際に前回と同じ値 (0x1000000) で試してみたところ、圧縮率は悪くなりました。そこで、今回は MIN_RANGE を 0x10000 に設定することにします。このため、出現頻度表の合計が MIN_RANGE 未満になるように count の値を調整します。
最初に、count の合計値を求め変数 m にセットします。m が MIN_RANGE 未満になるまで m を右へシフトしていき、その回数を変数 n にセットします。あとは count の各要素を n ビット右へシフトするだけです。このとき、出現頻度が 0 にならないように最下位ビットを 1 にしています。合計値が MIN_RANGE (0x10000) 未満になるので、記号の出現頻度も 2 バイトに収まります。
次は記号を符号化するメソッド encode() を作ります。
リスト : 符号化
def encode(self, rc, c):
temp = rc.range_ // self.count_sum[self.size]
rc.low += self.count_sum[c] * temp
rc.range_ = self.count[c] * temp
rc.encode_normalize()
encode() は Freq のメソッドで、前回のプログラムと同じです。桁上がりのないレンジコーダの場合、range_ と low の値を更新 (正規化) する処理がポイントになります。次のリストを見てください。
リスト : 符号化の正規化
def encode_normalize(self):
# range_ のチェック処理 1
while self.low & MASK1 == (self.low + self.range_) & MASK1:
self.stream.putc(self.low >> SHIFT)
self.low = (self.low << 8) & MASK
self.range_ <<= 8
# range_ のチェック処理 2
while self.range_ < MIN_RANGE:
self.range_ = (MIN_RANGE - (self.low & (MIN_RANGE - 1))) << 8
self.stream.putc(self.low >> SHIFT)
self.low = (self.low << 8) & MASK
range_ をチェックする処理 1 と 2 が「桁上がりのないレンジコーダ」を理解するポイントです。処理 1 では、low の上位 8 ビットの値と low + range_ の上位 8 ビットの値が同じかチェックしています。range_ の幅が狭くなると、low に range_ を加算しても上位 8 ビットの値は low と同じになります。たとえば、low の値が 0x10000000 で range_ の値が 0x100000 の場合を考えてみましょう。
low = 0x10000000 range_ = 0x00100000 -------------------------- low + range_ = 0x10100000
このように、low + range_ の上位 8 ビットは low と同じ 0x10 になります。つまり、この処理は range_ の値が小さくなったかチェックしているのです。そうであれば、range_ と low の値を 256 倍して low の上位 8 ビットを符号語として出力します。これは今までのレンジコーダと同じ処理です。
ところで、range_ の値が小さくなっても、low + range_ と low の上位 8 ビットの値が異なる場合があります。たとえば、low の値が 0x10ffa000 で range_ が 0xc000 の場合を考えてみます。
low = 0x10ffa000 range_ = 0x0000c000 -------------------------- low + range_ = 0x11006000
low の上位 8 ビットは 0x10 ですが、low + range_ の上位 8 ビットは 0x11 になります。ここで単純に low と range_ の値を 256 倍してみましょう。すると、low = 0xffa00000 と range_ = 0xc00000 になりますね。次に記号 c を読み込んだとき、low の増分が 0x600000 以上になると桁上がりが発生することになります。
low = low + range * count_sum[c] / count_sum[size] 0x_ffa00000 + 0x_00600000 -------------- 0x100000000 桁上がり！
つまり、low + range_ と low の上位 8 ビットの値が異なり、なおかつ range_ の幅が小さい場合は、桁上がりの可能性があるわけです。処理 1 で単純に range_ と定数値を比較しないのは、桁上がりの可能性をチェックするためなのです。この場合、処理 2 で range_ の値を修正します。
処理 2 では、range_ が MIN_RANGE (0x10000) より小さければ、桁上がりの可能性があるので range_ の値を修正します。たとえば、low の値が 0x10ffa000 の場合、range_ の値が 0x6000 以下であれば桁上がりは発生しません。この値は low の下位 16 ビットを取り出して、それを MIN_RANGE から引き算すれば求めることができます。次の式を見てください。
0x10000 - (0x10ffa0000 & (0x10000 - 1)) = 0x10000 - (0x10ffa0000 & 0xffff) = 0x10000 - 0xa000 = 0x6000
あとは、range_ と low の値を 256 倍して low の上位 8 ビットを符号語として出力します。
M.Hiroi は「桁上がりのないレンジコーダ」を参考文献『データ圧縮の基礎から応用まで』で知りましたが、このような簡単な方法で桁上がりを回避できることに大変驚きました。その反面、ずいぶん強引な方法だな、とも思いました。これで本当に復号できるのか疑問に思われる方もいるでしょう。実際、復号は今までのレンジコーダと違って、ちょっとだけ工夫が必要になります。
次は復号を行う Freq のメソッド decode() を作ります。
リスト : 復号
def decode(self, rc):
# 記号の探索
def search_code(value):
i = 0
j = self.size - 1
while i < j:
k = (i + j) // 2
if self.count_sum[k + 1] <= value:
i = k + 1
else:
j = k
return i
#
temp = rc.range_ // self.count_sum[self.size]
c = search_code((rc.code - rc.low) // temp)
rc.low += temp * self.count_sum[c]
rc.range_ = temp * self.count[c]
rc.decode_normalize()
return c
前回説明したレンジコーダの場合、ファイルから読み込んだ符号語は low に加算して、幅 range_ を狭めたら low の値を減算し、その値を使って復号しました。ところが、桁上がりのないレンジコーダの場合、range_ と low の両方の値を使って range_ を拡大するタイミングを決めています。したがって、復号の処理でも low を符号化と同じ値に再現できないと、符号語を正しく復号することはできません。
そこで、low の値は符号化と同じように計算して、その値を保持することにします。符号語を表す値は変数 code に保持します。最初に low は 0 に初期化し、code はファイルから 4 バイト読み込んで初期化します。この処理は RangeCoder のメソッド __init__() で行っています。
ここで、記号を表す値は code - low で求めることができる、ということに注意してください。ここがこのプログラムのポイントです。low の値は符号化と同様に増加していくので、code - low の値は減少していきます。この値が今までのレンジコーダの low に対応するわけです。局所関数 search_code() で記号を探索するときは (code - low) / temp を渡していることに注意してください。
そして、符号化と同じタイミングで range_ を拡大すれば、low の値は符号化と同様に再現することができます。次のリストを見てください。
リスト : 復号の正規化
def decode_normalize(self):
# range のチェック処理 1
while self.low & MASK1 == (self.low + self.range_) & MASK1:
self.code = ((self.code << 8) & MASK) + self.stream.getc()
self.low = (self.low << 8) & MASK
self.range_ <<= 8
# range のチェック処理 2
while self.range_ < MIN_RANGE:
self.range_ = (MIN_RANGE - (self.low & (MIN_RANGE - 1))) << 8
self.code = ((self.code << 8) & MASK) + self.stream.getc()
self.low = (self.low << 8) & MASK
復号で range_ と low の値を更新する場合、range_ のチェック条件は符号化と同じです。low と range_ の更新処理も符号化と同じですが、code の値を更新する処理を追加します。この処理は code を 256 倍してファイルから 1 バイト読み込んで加算するだけです。
あとのプログラムは簡単なので説明は割愛いたします。詳細は下記プログラムリストをお読みください。
それでは、実際に Canterbury Corpus で配布されているテストデータ The Canterbury Corpus を圧縮してみましょう。結果は次にようになりました。
表 : 桁上がりのないレンジコーダの結果
ファイル名 サイズ RangeCoder 下限値 符号化 復号
------------------------------------------------------------
alice29.txt 152,089 87,412 86,837 0.294 0.498
asyoulik.txt 125,179 75,820 75,235 0.250 0.418
cp.html 24,603 16,606 16,082 0.050 0.093
fields.c 11,150 7,501 6,980 0.027 0.037
grammar.lsp 3,721 2,674 2,155 0.008 0.014
kennedy.xls 1,029,744 461,022 459,971 1.885 3.318
lcet10.txt 426,754 249,796 249,071 0.811 1.363
plrabn12.txt 481,861 273,709 272,936 0.915 1.590
ptt5 513,216 78,346 77,636 0.798 1.397
sum 38,240 26,005 25,473 0.075 0.147
xargs.1 4,227 3,108 2,589 0.010 0.021
------------------------------------------------------------
合計 2,810,784 1,281,999 1,274,965 5.123 8.896
符号化と復号の単位 : 秒
実行環境 : Python 3.8.10, Ubuntu 20.04 LTS (WSL1), Intel Core i5-6200U 2.30GHz
圧縮率は、桁上がりがあるレンジコーダよりも少し悪くなる場合もありますが、それでも圧縮の限界に近い値となりました。桁上がりのないレンジコーダでも、高い性能を実現していることがわかります。実行時間ですが、桁上がりのあるレンジコーダと比べると、符号化は少し速くなりましたが、逆に復号は少し遅くなりました。
なお、実行時間の結果は M.Hiroi のコーディング、実行したマシン、プログラミング言語などの環境に大きく依存しています。また、これらの環境だけではなく、データの種類によっても実行時間はかなり左右されます。興味のある方は、いろいろなデータをご自分の環境で試してみてください。
今まで説明したハフマン符号やレンジコーダは静的符号化 (static coding) といい、あらかじめ記号の出現確率を調べておいて、それに基づいて入力記号列を符号化していく方法です。この方法では、ハフマン符号がもっとも有名でしょう。
これに対し、動的符号化 (dynamic coding) は入力記号列の符号化を行いながら記号の出現確率を変化させる方法で、適応型符号化 (adaptive coding) と呼ばれることもあります。最初は、どの記号も同じ確率で出現すると仮定して、記号列を読み込みながら記号の出現確率を修正し、その時点での出現確率に基づいて記号の符号化を行います。なお、辞書法の LZ 符号も動的符号化の一つです。
動的符号化の特徴は入力記号列の性質 (出現確率) の変化に適応できることですが、このほかにも長所があります。静的符号化の場合、復号するときに符号化で用いた記号の出現確率が必要になります。このため、レンジコーダのプログラムでは、記号の出現頻度表を出力ファイルの先頭に付加しています。ところが、動的符号化では復号しながら記号の出現確率を求めることができるので、出現頻度表をファイルに付加する必要はありません。
また、静的符号化でファイルを圧縮する場合、記号の出現頻度を求めるときにファイルからデータを読み込み、符号化を行うときに再度ファイルからデータを読み込む必要があります。このようにデータの入力が 2 回必要な圧縮アルゴリズムを 2 パスの圧縮アルゴリズムといいます。動的符号化は 1 パスで済むので、オンラインでのデータ圧縮にも対応することができます。
このように、動的符号化には有利な点があるため、ハフマン符号を動的符号化に対応させた適応型ハフマン符号が考案されています。しかしながら、適応型ハフマン符号は実装方法が難しく、処理速度も遅いという欠点があります。これに対し、適応型算術符号 (レンジコーダ) は簡単な方法で実装することができ、適応型レンジコーダは処理速度もそれほど遅くありません。とても優れた実装方法なのです。
今回は適応型レンジコーダを説明して、実際にファイルを圧縮してみましょう。
適応型レンジコーダの場合、出現頻度表をファイルに付加する必要がないため、記号の出現頻度を 2 バイトに丸める必要はありません。これは大きな利点です。あとは、累積度数の合計値が幅 range_ の最小値 MIN_RANGE より大きくならないように、出現頻度表を調整するだけです。この処理は、累積度数の合計値が MIN_RANGE に達したときに、各記号の出現頻度を半分にすることで実現できます。
それではプログラムを作りましょう。今回は桁上がりのあるレンジコーダを使います。次のリストを見てください。
リスト : 出現頻度表
class Freq:
def __init__(self, size):
self.size = size
self.count = [1] * size
self.sum_ = size
# 出現頻度表の更新
def update(self, c):
self.count[c] += 1
self.sum_ += 1
if self.sum_ >= MIN_RANGE:
n = 0
for x in range(self.size):
self.count[x] = (self.count[x] >> 1) | 1
n += self.count[x]
self.sum_ = n
Freq のインスタンス変数 count が記号の出現頻度表、size が出現する記号の種類、sum_ が count の合計値です。前回のように累積度数表を用意すると、その更新処理に時間がかかるようになります。そこで、記号の累積度数は計算で求めることにします。ただし、この処理に時間がかかると、適応型レンジコーダの実行速度は遅くなります。まずは簡単な方法でプログラムを作成し、実行速度が遅い場合は他の方法を試してみましょう。
適応型符号化の場合、最初はどの記号も同じ確率で出現すると仮定するのが一般的なので、出現する可能性がある記号はすべて count の要素を 1 に初期化します。そして、count の合計値 sum_ に size をセットします。
メソッド update() は記号 c の出現頻度を更新します。count[c] を +1 するとともに sum_ の値も +1 します。sum_ が MIN_RANGE 以上になったならば、各記号の出現頻度を 1 / 2 に更新します。このとき、出現頻度の値が 0 にならないように注意してください。プログラムでは最下位ビットを無条件で 1 にしています。
次は符号化を行うメソッド encode() を作ります。プログラムは次のようになります。
リスト : 記号の符号化
# 記号の累積度数を求める
def cumul(self, c):
n = 0
for x in range(c): n += self.count[x]
return n
# 符号化
def encode(self, rc, c):
temp = rc.range_ // self.sum_
rc.low += self.cumul(c) * temp
rc.range_ = self.count[c] * temp
rc.encode_normalize()
self.update(c)
記号 c の累積度数はメソッド cumul() で求めます。ここでは 0 から c までの記号数を加算して求めています。あとの処理は前回のプログラムとほぼ同じです。最後にメソッド update() を呼び出して出現頻度表を更新します。
最後に適応型レンジコーダで符号化を行う関数 encode() を作ります。
リスト : 適応型レンジコーダによる符号化
def encode(fin, fout):def encode(fin, fout):
rc = RangeCoder(fout, ENCODE)
freq = Freq(256)
for x in read_file(fin):
freq.encode(rc, x)
rc.finish()
Freq と RangeCoder のオブジェクトを生成して、変数 freq と rc にセットします。適応型符号化なので、ファイルを 2 回リードする必要はありません。read_file() で記号を読み込み、freq.encode() で記号を符号化するだけです。
次は復号を行うメソッド decode() を作ります。次のリストを見てください。
リスト : 記号の復号
def decode(self, rc):
# 記号の探索
def search_code(value):
n = 0
for c in range(self.size):
if value < n + self.count[c]: return c, n
n += self.count[c]
#
temp = rc.range_ // self.sum_
c, num = search_code(rc.low // temp)
rc.low -= temp * num
rc.range_ = temp * self.count[c]
rc.decode_normalize()
self.update(c)
return c
累積度数表がないので、記号の探索を行う関数 search_code() で二分探索は使えません。今回は単純な線形探索で記号を求めています。この処理は時間がかかるので、復号は相当に遅くなることが予想されます。あとは、前回のプログラムとほぼ同じですが、符号化と同じくメソッド update() を呼び出して、記号の出現頻度表を更新することを忘れないでください。
最後に、適応型レンジコーダで復号を行う関数 decode() を作ります。
リスト : 適応型レンジコーダによる復号
def decode(fin, fout, size):
freq = Freq(256)
rc = RangeCoder(fin, DECODE)
for _ in range(size):
fout.putc(freq.decode(rc))
Freq と RangeCoder のオブジェクトを生成して、変数 freq と rc にセットします。あとは size 個の記号を freq.decode() で復号するだけです。
なお、今回は記号の種類を 256 (0 - 255) としましたが、終端記号 (256 : END) を含めて 257 種類とする方法もあります。この場合は、元のファイルサイズをファイルに付加する必要はありません。符号化のときは最後に END を符号化し、復号のときは END を復号した時点で処理を終了するようにします。興味のある方は試してみてください。
それでは、実際に Canterbury Corpus で配布されているテストデータ The Canterbury Corpus を圧縮してみましょう。結果は次にようになりました。
表 : 適応型レンジコーダの結果
() は ファイルサイズと出現頻度表を除いた値
ファイル名 サイズ ARC 下限値 符号化 復号
-----------------------------------------------------------
alice29.txt 152,089 87,147 86,837 1.219 2.239
asyoulik.txt 125,179 75,533 75,235 1.045 1.839
cp.html 24,603 16,299 16,082 0.197 0.363
fields.c 11,150 7,164 6,980 0.083 0.149
grammar.lsp 3,721 2,305 2,155 0.026 0.051
kennedy.xls 1,029,744 460,734 459,971 3.469 5.096
lcet10.txt 426,754 249,491 249,071 3.574 6.589
plrabn12.txt 481,861 273,392 272,936 3.968 7.060
ptt5 513,216 78,090 77,636 1.458 2.289
sum 38,240 25,638 25,473 0.239 0.381
xargs.1 4,227 2,743 2,589 0.033 0.070
-----------------------------------------------------------
合計 2,810,784 1,278,536 1,274,965 15.131 26.126
符号化と復号の単位 : 秒
適応型レンジコーダの圧縮率は、静的なレンジコーダと同様に圧縮の限界に近い値になりました。出現頻度表を付加しない分だけ、多くのファイルで静的なレンジコーダよりも高い圧縮率になりましたが、小さなファイルは逆に圧縮率が悪くなるようです。
適応型符号化の場合、出現しない記号が多数あると、圧縮率が少し悪くなるという欠点があります。たとえば、記号が 0 と 1 しかないデータを符号化してみましょう。適応型レンジコーダでは記号 0 - 255 の出現頻度を 1 に初期化しています。このため、記号数が少ないうちは記号 2 - 255 の出現頻度の影響が大きくなり、圧縮率はどうしても悪くなってしまいます。
ようするに、記号をたくさん読み込まないと、その出現頻度表の確率はあてにならないというわけです。したがって、小さなファイルの圧縮率は静的なレンジコーダよりも悪くなる場合が多いようです。逆に、大きなファイルであれば、静的なレンジコーダと同様に高い圧縮率を達成することができます。
実行時間ですが、符号化と復号ともに静的なレンジコーダよりも相当に遅くなりました。やっぱり、適応型レンジコーダは時間がかかりますね。とくに復号は 3 - 4 倍程度の時間がかかっています。復号は記号の探索に線形探索を使っているので、時間がかかるのは想定内の結果といえるでしょう。
なお、実行時間の結果は M.Hiroi のコーディング、実行したマシン、プログラミング言語などの環境に大きく依存しています。また、これらの環境だけではなく、データの種類によっても実行時間はかなり左右されます。興味のある方は、いろいろなデータをご自分の環境で試してみてください。
#
# rangecoder1.py : 桁上がりのないレンジコーダ (Range Coder)
#
# Copyright (C) 2007-2022 Makoto Hiroi
#
from byteio import *
# 定数
ENCODE = "encode"
DECODE = "decode"
MAX_RANGE = 0xffffffff # 修正 (2007/10/06)
MIN_RANGE = 0x10000
MASK = 0xffffffff
MASK1 = 0xff000000
SHIFT = 24
# 桁上がりのないレンジコーダ
class RangeCoder:
def __init__(self, s, mode):
self.stream = s
self.range_ = MAX_RANGE
self.low = 0
if mode == DECODE:
# 4 byte read
self.code = self.stream.getc()
self.code = (self.code << 8) + self.stream.getc()
self.code = (self.code << 8) + self.stream.getc()
self.code = (self.code << 8) + self.stream.getc()
elif mode != ENCODE:
raise "RangeCoder mode error"
# 符号化の正規化
def encode_normalize(self):
while self.low & MASK1 == (self.low + self.range_) & MASK1:
self.stream.putc(self.low >> SHIFT)
self.low = (self.low << 8) & MASK
self.range_ <<= 8
while self.range_ < MIN_RANGE:
self.range_ = (MIN_RANGE - (self.low & (MIN_RANGE - 1))) << 8
self.stream.putc(self.low >> SHIFT)
self.low = (self.low << 8) & MASK
# 復号の正規化
def decode_normalize(self):
while self.low & MASK1 == (self.low + self.range_) & MASK1:
self.code = ((self.code << 8) & MASK) + self.stream.getc()
self.low = (self.low << 8) & MASK
self.range_ <<= 8
while self.range_ < MIN_RANGE:
self.range_ = (MIN_RANGE - (self.low & (MIN_RANGE - 1))) << 8
self.code = ((self.code << 8) & MASK) + self.stream.getc()
self.low = (self.low << 8) & MASK
# 終了
def finish(self):
self.stream.putc((self.low >> 24) & 0xff)
self.stream.putc((self.low >> 16) & 0xff)
self.stream.putc((self.low >> 8) & 0xff)
self.stream.putc(self.low & 0xff)
#
# rc01.py : 静的なレンジコーダ
#
# Copyright (C) 2007-2022 Makoto Hiroi
#
import argparse, os.path, time
from byteio import *
# 桁上がりのないレンジコーダ
from rangecoder1 import *
# 出現頻度表
class Freq:
def __init__(self, count):
size = len(count)
self.size = size
m = sum(count)
# 正規化
if m >= MIN_RANGE:
self.count = [0] * size
n = 0
while m >= MIN_RANGE:
m >>= 1
n += 1
for x in range(size):
if count[x] != 0:
self.count[x] = (count[x] >> n) | 1
else:
self.count = count[:]
self.count_sum = [0] * (size + 1)
# 累積度数表
for x in range(size):
self.count_sum[x + 1] = self.count_sum[x] + self.count[x]
# 出現頻度表の出力
def write_count_table(self, fout):
for x in self.count:
fout.putc(x >> 8)
fout.putc(x & 0xff)
# 符号化
def encode(self, rc, c):
temp = rc.range_ // self.count_sum[self.size]
rc.low += self.count_sum[c] * temp
rc.range_ = self.count[c] * temp
rc.encode_normalize()
# 復号
def decode(self, rc):
# 記号の探索
def search_code(value):
i = 0
j = self.size - 1
while i < j:
k = (i + j) // 2
if self.count_sum[k + 1] <= value:
i = k + 1
else:
j = k
return i
#
temp = rc.range_ // self.count_sum[self.size]
c = search_code((rc.code - rc.low) // temp)
rc.low += temp * self.count_sum[c]
rc.range_ = temp * self.count[c]
rc.decode_normalize()
return c
# ファイルの読み込み
def read_file(fin):
while True:
c = fin.getc()
if c is None: break
yield c
# レンジコーダによる符号化
def encode(fin, fout):
count = [0] * 256
for x in read_file(fin):
count[x] += 1
rc = RangeCoder(fout, ENCODE)
freq = Freq(count)
freq.write_count_table(fout)
fin.stream.seek(0)
for x in read_file(fin):
freq.encode(rc, x)
rc.finish()
# 出現頻度表の読み込み
def read_count_table(fin):
count = [0] * 256
for x in range(256):
count[x] = (fin.getc() << 8) + fin.getc()
return count
# レンジコーダによる復号
def decode(fin, fout, size):
freq = Freq(read_count_table(fin))
rc = RangeCoder(fin, DECODE)
for _ in range(size):
fout.putc(freq.decode(rc))
# 符号化
def encode_file(name1, name2):
size = os.path.getsize(name1)
with ByteIO(name1, ROPEN) as fin, ByteIO(name2, WOPEN) as fout:
fout.putc((size >> 24) & 0xff)
fout.putc((size >> 16) & 0xff)
fout.putc((size >> 8) & 0xff)
fout.putc(size & 0xff)
if size > 0: encode(fin, fout)
# 復号
def decode_file(name1, name2):
with ByteIO(name1, ROPEN) as fin, ByteIO(name2, WOPEN) as fout:
size = 0
for _ in range(4):
size = (size << 8) + fin.getc()
if size > 0: decode(fin, fout, size)
# オプション解析
parser = argparse.ArgumentParser(description='RangeCoder符号')
parser.add_argument('name1', help='入力ファイル')
parser.add_argument('name2', help='出力ファイル')
parser.add_argument('-e', '--encode', action='store_true', help='符号化')
parser.add_argument('-d', '--decode', action='store_true', help='復号')
args = parser.parse_args()
# 実行
s = time.time()
if args.encode and not args.decode:
encode_file(args.name1, args.name2)
elif args.decode:
decode_file(args.name1, args.name2)
else:
print('option error')
e = time.time()
print('{:.3f}'.format(e - s))
#
# rc1.py : 適応型レンジコーダ
#
# Copyright (C) 2007-2022 Makoto Hiroi
#
import argparse, os.path, time
from rangecoder import *
# 出現頻度表
class Freq:
def __init__(self, size):
self.size = size
self.count = [1] * size
self.sum_ = size
# 出現頻度表の更新
def update(self, c):
self.count[c] += 1
self.sum_ += 1
if self.sum_ >= MIN_RANGE:
n = 0
for x in range(self.size):
self.count[x] = (self.count[x] >> 1) | 1
n += self.count[x]
self.sum_ = n
# 記号の累積度数を求める
def cumul(self, c):
n = 0
for x in range(c): n += self.count[x]
return n
# 符号化
def encode(self, rc, c):
temp = rc.range_ // self.sum_
rc.low += self.cumul(c) * temp
rc.range_ = self.count[c] * temp
rc.encode_normalize()
self.update(c)
# 復号
def decode(self, rc):
# 記号の探索
def search_code(value):
n = 0
for c in range(self.size):
if value < n + self.count[c]: return c, n
n += self.count[c]
#
temp = rc.range_ // self.sum_
c, num = search_code(rc.low // temp)
rc.low -= temp * num
rc.range_ = temp * self.count[c]
rc.decode_normalize()
self.update(c)
return c
# ファイルの読み込み
def read_file(fin):
while True:
c = fin.getc()
if c is None: break
yield c
# レンジコーダによる符号化
def encode(fin, fout):
rc = RangeCoder(fout, ENCODE)
freq = Freq(256)
for x in read_file(fin):
freq.encode(rc, x)
rc.finish()
# レンジコーダによる復号
def decode(fin, fout, size):
freq = Freq(256)
rc = RangeCoder(fin, DECODE)
for _ in range(size):
fout.putc(freq.decode(rc))
# 符号化
def encode_file(name1, name2):
size = os.path.getsize(name1)
with ByteIO(name1, ROPEN) as fin, ByteIO(name2, WOPEN) as fout:
fout.putc((size >> 24) & 0xff)
fout.putc((size >> 16) & 0xff)
fout.putc((size >> 8) & 0xff)
fout.putc(size & 0xff)
if size > 0: encode(fin, fout)
# 復号
def decode_file(name1, name2):
with ByteIO(name1, ROPEN) as fin, ByteIO(name2, WOPEN) as fout:
size = 0
for _ in range(4):
size = (size << 8) + fin.getc()
if size > 0: decode(fin, fout, size)
# オプション解析
parser = argparse.ArgumentParser(description='適応型RangeCoder符号')
parser.add_argument('name1', help='入力ファイル')
parser.add_argument('name2', help='出力ファイル')
parser.add_argument('-e', '--encode', action='store_true', help='符号化')
parser.add_argument('-d', '--decode', action='store_true', help='復号')
args = parser.parse_args()
# 実行
s = time.time()
if args.encode and not args.decode:
encode_file(args.name1, args.name2)
elif args.decode:
decode_file(args.name1, args.name2)
else:
print('option error')
e = time.time()
print('{:.3f}'.format(e - s))