如何在Python3中快速遍历二进制文件比特并优化匹配性能？

我需要在Python3中快速遍历二进制文件的每一个比特：读取所有字节并转换为比特流，逐比特更新移位寄存器，当移位寄存器的值与同步标记的汉明距离≤3时执行指定函数。但以下代码运行缓慢，求优化方案：

import Levenshtein, argparse, io
parser=argparse.ArgumentParser()
parser.add_argument("-o", "--output", help="Output binary file name")
parser.add_argument("-i", "--input", help="Input binary file name")
inputfile=parser.parse_args().input
outfile=parser.parse_args().output

def get_byte(f):
    while(byte := f.read(1)):
        return bin(int().from_bytes(byte, 'big'))[2:].zfill(8)

def func():
    # text
    return

def main(input_file, sync_marker, sync_buffer, out, out_filename):
    bit_array=''
    k=0
    while True:
        if(len(bit_array)<=1):
            bits_array=get_byte(f=input_file)
            k=0
        if(bits_array!=None):
            bit_array=bits_array[k:]
            k+=1
            sync_buffer=sync_buffer[1:]+bit_array[:1]
            err=int(Levenshtein.hamming(sync_buffer, str(sync_marker)))
            if(err<int(3)):
                func()
        if(bits_array==None):
            with open(out_filename, 'wb') as out_bytes_file:
                out_bytes_file.write(out.getvalue())
            input_file.close()
            break
    return

if(__name__=='__main__'):
    f=open(inputfile, "rb")
    out_file=io.BytesIO()
    sync_marker='0100101001'
    sync_buffer=str('0'*int(len(sync_marker)))
    main(input_file=f, sync_buffer=sync_buffer, sync_marker=sync_marker, out=out_file, out_filename=outfile)

优化思路 #

1. 避免逐字节IO，批量读取文件 #

原代码用read(1)逐字节读取，IO开销极大。改为分块读取（比如每次读1MB），大幅减少系统调用次数。

2. 用整数/位运算替代字符串处理比特 #

字符串是不可变类型，每次切片、拼接都会生成新对象，性能很差。改用整数表示移位寄存器和同步标记，用位运算实现快速更新和比较：

• 将同步标记转为整数，预计算掩码保证移位寄存器长度固定
• 移位寄存器更新：sync_buffer = (sync_buffer << 1 | new_bit) & mask
• 汉明距离用位异或+统计1的个数实现：bin(sync_buffer ^ sync_marker_int).count('1')，比调用Levenshtein库快一个数量级

3. 直接从字节提取比特，跳过字符串转换 #

无需将字节转为二进制字符串再切片，直接用位运算从字节中逐位提取比特（从高位到低位）。

4. 简化逻辑，消除冗余操作 #

原代码中bit_array和k的逻辑混乱，改为逐字节处理所有8个比特，流程更清晰，减少不必要的变量操作。

优化后的代码 #

import argparse, io

def func():
    # 这里写你的业务逻辑
    pass

def main(input_filename, sync_marker, out_filename):
    sync_len = len(sync_marker)
    sync_marker_int = int(sync_marker, 2)
    mask = (1 << sync_len) - 1  # 用于保持移位寄存器长度
    sync_buffer = 0  # 用整数表示移位寄存器，初始全0

    out = io.BytesIO()

    # 批量读取文件，每次处理1MB块
    chunk_size = 1024 * 1024
    with open(input_filename, "rb") as f:
        while chunk := f.read(chunk_size):
            for byte in chunk:
                # 从高位到低位提取当前字节的8个比特
                for i in reversed(range(8)):
                    bit = (byte >> i) & 1
                    # 更新移位寄存器
                    sync_buffer = (sync_buffer << 1 | bit) & mask
                    # 计算汉明距离
                    hamming_dist = bin(sync_buffer ^ sync_marker_int).count('1')
                    if hamming_dist <= 3:
                        func()

    # 写入输出文件
    with open(out_filename, 'wb') as out_file:
        out_file.write(out.getvalue())

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument("-o", "--output", required=True, help="Output binary file name")
    parser.add_argument("-i", "--input", required=True, help="Input binary file name")
    args = parser.parse_args()

    sync_marker = '0100101001'
    main(args.input, sync_marker, args.output)

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