AB编码器程序入门指南：从原理到实战避坑

1次阅读

共计 1964 个字符，预计需要花费 5 分钟才能阅读完成。

传统编码方案（如 ASCII 或定长编码）在数据压缩和传输中存在明显缺陷：

空间效率低 ：无法根据符号出现频率动态调整编码长度，高频符号与低频符号占用相同存储空间
冗余度高 ：对重复模式（如连续相同字符）缺乏有效压缩手段
实时性差 ：流式数据处理时需要完整加载内容后才能开始编码

编码类型	时间复杂度	空间复杂度	最佳适用场景
Huffman	O(nlogn)	O(k)	静态已知概率分布的离散数据
Arithmetic	O(n)	O(1)	高精度概率建模场景
AB 编码器	O(n)	O(m)	实时流数据 / 动态概率分布

关键差异：

AB 编码器采用自适应概率模型，无需预先统计符号频率
通过分块处理实现 O(1) 的单符号编码延迟
支持动态调整编码表，适合网络传输场景

1. 初始化符号概率表为均匀分布
2. 对输入数据分块（建议 4KB~16KB）：a. 计算当前块符号频率分布
   b. 构建局部霍夫曼树
   c. 编码块头部信息（符号表 + 树结构）d. 按树结构进行符号编码
3. 输出块校验码（CRC32）

import heapq
from collections import defaultdict

class ABEncoder:
    def __init__(self, chunk_size=4096):
        self.chunk_size = chunk_size

    def encode(self, data):
        """时间复杂度 O(n)，空间复杂度 O(m) 其中 m 为符号表大小"""
        chunks = [data[i:i+self.chunk_size] 
                 for i in range(0, len(data), self.chunk_size)]

        encoded = bytearray()
        for chunk in chunks:
            # 动态构建频率表
            freq = defaultdict(int)
            for sym in chunk:
                freq[sym] += 1

            # 构建霍夫曼树
            heap = [[weight, [sym, ""]] for sym, weight in freq.items()]
            heapq.heapify(heap)
            while len(heap) > 1:
                lo = heapq.heappop(heap)
                hi = heapq.heappop(heap)
                for pair in lo[1:]:
                    pair[1] = '0' + pair[1]
                for pair in hi[1:]:
                    pair[1] = '1' + pair[1]
                heapq.heappush(heap, [lo[0] + hi[0]] + lo[1:] + hi[1:])

            # 生成编码表
            huff_table = {sym: code for sym, code in heap[0][1:]}

            # 写入块头（符号表）encoded.extend(len(huff_table).to_bytes(2, 'big'))
            for sym, code in huff_table.items():
                encoded.extend([sym, len(code)])
                encoded.extend(int(code, 2).to_bytes((len(code)+7)//8, 'big'))

            # 写入编码数据
            bits = ''.join(huff_table[sym] for sym in chunk)
            padding = 8 - len(bits) % 8
            bits += '0' * padding
            encoded.extend(padding.to_bytes(1, 'big'))
            for i in range(0, len(bits), 8):
                encoded.append(int(bits[i:i+8], 2))

        return bytes(encoded)