Python实现极限Bitpacking压缩比特流、降低存储占用方案问询

问题背景 #

需要将多组离散状态变量尽可能紧凑地打包进比特流，现有状态数量列表如下：

Number_of_states = [3,5,129,15,6,2] # 实际场景中该列表长度更长

当前采用位域打包方案，占用比特数为2+3+8+4+3+1=21bit，但理论最小存储比特为np.log2(3*5*129*15*6*2)=18.4bit，可节省2bit。实际场景中当前打包总长度为298bit，达到理论压缩效率可节省5%以上的空间，对业务收益明显。
之前尝试过packalgorithms实现，但小数据量场景下额外开销过高。由于打包格式字符串为固定值，会预先传输，不存在额外开销问题。

现有实现代码 #

from bitstring import pack
import numpy as np

DATA_TO_BE_PACKED=np.zeros(6)
Number_of_states=[3,5,129,15,6,2]# 实际场景中长度更长
DATA_TO_BE_PACKED=np.random.randint(Number_of_states)
string=''
for item in Number_of_states:
    string+='uint:{}, '.format(int(np.ceil(np.log2(item))))
PACKED_DATA = pack(string,*DATA_TO_BE_PACKED)
print(len(PACKED_DATA ))
print(PACKED_DATA.unpack(string))

可行实现方案 #

你要的极限打包本质是基数混合编码，把所有状态值转换成以各状态数为基数的大整数，再把这个大整数转成二进制比特流即可，解压时做逆运算就行，完全没有额外开销，正好适配你格式预先约定的场景。

实现逻辑 #

• 预先计算每个维度的累积乘积（步长），比如第一个维度步长是1，第二个是3，第三个是3*5=15，以此类推
• 把每个状态值乘对应步长求和，得到一个唯一的整数
• 把这个整数转成比特流，占用比特数就是理论上的ceil(log2(总状态数))，刚好达到极限压缩率

完整代码 #

import numpy as np
from bitstring import BitArray

# 预先计算固定参数，打包、解包端提前存储即可，不需要随比特流传输
def pre_calc_params(state_counts):
    steps = [1]
    for n in state_counts[:-1]:
        steps.append(steps[-1] * n)
    total_states = steps[-1] * state_counts[-1]
    total_bits = int(np.ceil(np.log2(total_states)))
    return steps, total_bits

# 打包函数
def pack_values(values, steps, total_bits):
    num = 0
    for v, step in zip(values, steps):
        num += v * step
    return BitArray(uint=num, length=total_bits)

# 解包函数
def unpack_values(bit_arr, steps, state_counts):
    num = bit_arr.uint
    values = []
    remaining = num
    for step, n in zip(reversed(steps), reversed(state_counts)):
        v = remaining // step
        values.append(v)
        remaining = remaining % step
    return values[::-1]

# 功能测试
if __name__ == "__main__":
    Number_of_states = [3,5,129,15,6,2]
    steps, total_bits = pre_calc_params(Number_of_states)
    print(f"总占用比特数：{total_bits}") # 输出19，刚好是ceil(18.4)的结果，比原方案少2bit
    
    # 随机生成测试数据
    DATA_TO_BE_PACKED = np.random.randint(Number_of_states)
    print(f"原始数据：{DATA_TO_BE_PACKED}")
    
    # 打包
    packed = pack_values(DATA_TO_BE_PACKED, steps, total_bits)
    print(f"打包后比特长度：{len(packed)}")
    
    # 解包
    unpacked = unpack_values(packed, steps, Number_of_states)
    print(f"解包后数据：{unpacked}")
    print(f"数据一致性校验：{np.array_equal(DATA_TO_BE_PACKED, unpacked)}")

方案优势 #

• 完全达到理论极限压缩率，没有任何冗余
• 无额外元数据开销，只要打包解包双方提前存好固定参数即可，完全符合你格式预先约定的场景
• 运算量极低，不管状态列表多长都能快速处理，小数据量场景也没有额外开销
• 支持超长状态列表，Python原生支持大整数运算，不需要额外处理溢出问题

内容的提问来源于stack exchange，提问作者Okapi575