一、问题描述和基本要求

对一幅BMP格式的灰度图像进行二元Fano编码、译码。

二、算法思想

1949年，费诺(R.M. Fano)提出了一种编码方法，称之为费诺码或Fano码。它属于概率匹配编码，但一般也不是最佳的编码方法，只有当信源的概率分布呈现分布形式的条件下，才能达到最佳码的性能。元费诺码的编码步骤如下：
① 将个信源符号按概率分布的大小，以递减的次序排列起来，设

F(start,end,possibility,string):
    if start == end:
        p_start对应的信源符号s_start的码字为string
        return
    找到符合上述条件的k_1和k_2，分别计算偏差值d_1和d_2，选择偏差值最小的k
    F(start,k,sum(p_start,p_{start+1},…,p_k ),string+"0")
        F(k+1,end,sum(p_{k+1},p_{k+2},…,p_end ),string+"1")

由上述递归结构F即可求得每个信源符号的码字，从而可以对文件进行编码和解码。

三、源代码

#!/usr/bin/env python3
import os
import pickle
from math import log
# python3 -m pip install matplotlib
import matplotlib.pyplot as plt
# python3 -m pip install numpy
import numpy as np
# python3 -m pip install opencv-python
from cv2 import IMREAD_GRAYSCALE, absdiff, imread, imshow, imwrite
# python3 -m pip install prettytable
from prettytable import PrettyTable

class WriterWrapper:
    def __init__(self, f):
        self.buf = ""
        self.f = f

    def __call__(self, con: str):
        self.buf += con
        while len(self.buf) >= 8:
            out = self.buf[0:8]
            self.buf = self.buf[8:]
            conint = int(out, 2)
            self.f.write(bytes([conint]))

    def flush(self):
        if len(self.buf) > 0:
            self.buf += "0"*(7 - ((len(self.buf) - 1) % 8))
        while len(self.buf) > 0:
            out = self.buf[0:8]
            self.buf = self.buf[8:]
            conint = int(out, 2)
            self.f.write(bytes([conint]))

class ReaderWrapper:
    def __init__(self, f):
        self.buf = ""
        self.f = f

    def next(self):
        if self.buf != "":
            c = self.buf[0]
            self.buf = self.buf[1:]
            return c
        c = self.f.read(1)
        if c == b"":
            return ""
        bstr = bin(ord(c))[2:]
        bstr = "0"*(7 - ((len(bstr) - 1) % 8)) + bstr
        self.buf = bstr
        return self.next()

    def read(self, c : int):
        s = ""
        for i in c:
            s += self.next()
        return s

class Symbol:
    def __init__(self, symbol, weight, pos):
        self.word = ""
        self.pos = pos
        self.weight = weight
        self.symbol = symbol
    
    def __lt__(self, value):
        return self.pos < value.pos

class FanoNode:
    def __init__(self, left = None, right = None):
        self.symbol = None
        self.left = left
        self.right = right

def Fano(start, end, possibility, string, syms) -> FanoNode:
    leaf = FanoNode()
    if start == end - 1:
        leaf.symbol = syms[start]
        syms[start].word = string
        return leaf
    mid = possibility / 2.0
    target = 0.0
    index = start
    while target < mid:
        target += syms[index].pos
        index += 1
    subseq1left = syms[start:index-1]
    subseq1right = syms[index-1:end]
    pos1left = sum(map(lambda x: x.pos, subseq1left))
    pos1right = sum(map(lambda x: x.pos, subseq1right))
    delta1 = abs(pos1left-pos1right)
    subseq2left = syms[start:index]
    subseq2right = syms[index:end]
    pos2left = sum(map(lambda x: x.pos, subseq2left))
    pos2right = sum(map(lambda x: x.pos, subseq2right))
    delta2 = abs(pos2left-pos2right)
    if delta1 < delta2:
        leaf.left = Fano(start, index-1, pos1left, string + "0", syms)
        leaf.right = Fano(index-1, end, pos1right, string + "1", syms)
    else:
        leaf.left = Fano(start, index, pos2left, string + "0", syms)
        leaf.right = Fano(index, end, pos2right, string + "1", syms)
    return leaf

if __name__ == "__main__":
    print("请将要压缩的图像文件命名为origin.bmp。")
    print("读入源图像文件...", end="")
    originimg = imread("origin.bmp", IMREAD_GRAYSCALE)
    originfilesize = os.path.getsize("origin.bmp")
    print("成功")
    print("分析原图像文件...", end="")
    originshape = originimg.shape
    originsize = originimg.size
    flatten = originimg.flatten()
    counter = dict.fromkeys(set(flatten), 0)
    table = PrettyTable()
    table.field_names = ["信源符号", "数量", "概率"]
    SymbolStringMap = {}
    syms = []
    for item in flatten:
        counter[item] += 1
    for item in counter:
        syms.append(Symbol(item, counter[item], counter[item] / originsize))
        table.add_row([item, counter[item], counter[item] / originsize])
    syms.sort(reverse=True)
    Hs = sum(map(lambda weight: -weight / originsize *
                 log(weight / originsize, 2), counter.values()))
    table.sortby = "数量"
    table.reversesort = True
    print("完成")
    print("原图像分辨率为：{1}*{0}，作为信源，共有{2}个、{3}种信源符号，熵为{4:.2f}, "
          "信源概率表如下：".format(*originshape, originsize, len(counter), Hs))
    print(table)
    print("正在进行费诺编码...")
    Fano(0, len(syms), 1.0, "", syms)
    table = PrettyTable()
    table.field_names = ["信源符号", "码字", "码字长度", "数量", "概率"]
    Len = 0
    for sym in syms:
        table.add_row([sym.symbol, sym.word, len(sym.word),
                       sym.weight, sym.pos])
        Len += sym.weight*len(sym.word)
        SymbolStringMap[sym.symbol] = sym.word
    print("编码完成，平均码长为{:.2f}，编码表如下：".format(Len/originsize))
    table.sortby = "数量"
    print(table)
    print("信源的熵为{:.2f}，平均码长为{:.2f}，编码效率为{:.2f}%".format(Hs, Len/originsize, Hs/Len*originsize*100))
    print("将编码后的文件写入到新文件target.bin中。编码采用费诺编码，先写入图像尺寸和文件长度，再写入编码表，最后写入编码后图像矩阵")
    f = open("target.bin", "wb")
    # 写入文件分辨率和长度
    f.write(pickle.dumps(originshape))
    f.write(pickle.dumps(Len))
    # 写入编码表
    f.write(pickle.dumps(counter))
    # 写入图像矩阵
    wt = WriterWrapper(f)
    for item in flatten:
        wt(SymbolStringMap[item])
    wt.flush()
    f.close()
    TarLen = os.path.getsize("target.bin")
    print("新文件target.bin的大小为{}字节，源文件大小为{}字节，压缩比为{:.2f}%".format(
        TarLen, originfilesize, TarLen/originfilesize*100))
    print("现在开始从新文件target.bin中恢复图像信息...")
    f = open("target.bin", "rb")
    # 读取图像分辨率
    originshape = pickle.load(f)
    # 读取长度
    Len = pickle.load(f)
    # 读取编码表
    counter = pickle.load(f)
    print("正在进行费诺译码...")
    syms = []
    originsize = originshape[0]*originshape[1]
    for item in counter:
        syms.append(Symbol(item, counter[item], counter[item] / originsize))
    syms.sort(reverse=True)
    p = root = Fano(0, len(syms), 1.0, "", syms)
    cur = 0
    rd = ReaderWrapper(f)
    imgflatten = []
    while (cur != Len):
        c = rd.next()
        cur += 1
        if c == "0":
            p = p.left
        elif c == "1":
            p = p.right
        else:
            raise AssertionError("c must be 0 or 1")
            break
        if (p.left == None) and (p.right == None):
            imgflatten.append(p.symbol.symbol)
            p = root
    imgarr = np.array(imgflatten, dtype=np.uint8)
    imgarr.resize(originshape)
    imwrite("target.bmp", imgarr)
    dif = absdiff(originimg, imgarr)
    plt.rcParams['font.sans-serif'] = ['SimHei']
    plt.rcParams['axes.unicode_minus'] = False
    fig = plt.figure('result')
    plt.subplot(1, 3, 1)
    plt.title('原图像')
    plt.imshow(originimg, plt.cm.gray)
    plt.axis('off')
    plt.subplot(1, 3, 2)
    plt.title('编解码后的图像')
    plt.imshow(imgarr, plt.cm.gray)
    plt.axis('off')
    plt.subplot(1, 3, 3)
    plt.title('差异')
    plt.imshow(dif, plt.cm.gray)
    plt.subplots_adjust(wspace=0.15)
    plt.axis('off')
    plt.show()