在Python中,我如何在二进制文件中读取并循环该文件的每个字节?


当前回答

要读取一个文件-一次一个字节(忽略缓冲)-你可以使用双参数iter(callable, sentinel)内置函数:

with open(filename, 'rb') as file:
    for byte in iter(lambda: file.read(1), b''):
        # Do stuff with byte

它调用file.read(1),直到没有返回b”(空字节串)。对于大文件,内存不会无限增长。你可以将buffering=0传递给open()来禁用缓冲——它保证每次迭代只读取一个字节(慢)。

With-statement自动关闭文件——包括下面的代码引发异常的情况。

尽管默认情况下存在内部缓冲,但一次处理一个字节的效率仍然很低。例如,下面是黑洞.py实用程序,它会吃掉所有给定的东西:

#!/usr/bin/env python3
"""Discard all input. `cat > /dev/null` analog."""
import sys
from functools import partial
from collections import deque

chunksize = int(sys.argv[1]) if len(sys.argv) > 1 else (1 << 15)
deque(iter(partial(sys.stdin.detach().read, chunksize), b''), maxlen=0)

例子:

$ dd if=/dev/zero bs=1M count=1000 | python3 blackhole.py

在我的机器上,当chunksize == 32768时,它处理大约1.5 GB/s,当chunksize == 1时,它只处理大约7.5 MB/s。也就是说,每次读取一个字节要慢200倍。考虑一下您是否可以重写处理以便一次使用多个字节,以及您是否需要性能。

Mmap允许您同时将文件视为bytearray和文件对象。如果需要访问两个接口,它可以作为在内存中加载整个文件的替代方案。特别是,你可以在一个内存映射文件上一次迭代一个字节,只使用简单的for循环:

from mmap import ACCESS_READ, mmap

with open(filename, 'rb', 0) as f, mmap(f.fileno(), 0, access=ACCESS_READ) as s:
    for byte in s: # length is equal to the current file size
        # Do stuff with byte

Mmap支持切片表示法。例如,mm[i:i+len]返回文件中从位置i开始的len字节。Python 3.2之前不支持上下文管理器协议;在这种情况下,需要显式调用mm.close()。使用mmap遍历每个字节比file.read(1)消耗更多的内存,但是mmap要快一个数量级。

其他回答

在尝试了以上所有方法并使用@Aaron Hall的答案后,我在一台运行windows 10, 8gb RAM和Python 3.5 32位的计算机上得到了一个~ 90mb的文件的内存错误。我的一位同事推荐我使用numpy,它的效果非常好。

到目前为止,读取整个二进制文件(我测试过)的最快速度是:

import numpy as np

file = "binary_file.bin"
data = np.fromfile(file, 'u1')

参考

比目前任何方法都要快。希望它能帮助到一些人!

总结chrispy, Skurmedel, Ben Hoyt和Peter Hansen的所有出色之处,这将是一次一个字节处理二进制文件的最佳解决方案:

with open("myfile", "rb") as f:
    while True:
        byte = f.read(1)
        if not byte:
            break
        do_stuff_with(ord(byte))

对于python 2.6及以上版本,因为:

Python内部缓冲区-不需要读取块 DRY原则——不重复读行 语句确保干净的文件关闭 当没有更多字节时,'byte'的计算结果为false(当字节为零时不是)

或使用J. F.塞巴斯蒂安的解决方案提高速度

from functools import partial

with open(filename, 'rb') as file:
    for byte in iter(partial(file.read, 1), b''):
        # Do stuff with byte

或者如果你想把它作为一个生成器函数,就像codeape演示的那样:

def bytes_from_file(filename):
    with open(filename, "rb") as f:
        while True:
            byte = f.read(1)
            if not byte:
                break
            yield(ord(byte))

# example:
for b in bytes_from_file('filename'):
    do_stuff_with(b)

这个生成器从文件中产生字节,以块的形式读取文件:

def bytes_from_file(filename, chunksize=8192):
    with open(filename, "rb") as f:
        while True:
            chunk = f.read(chunksize)
            if chunk:
                for b in chunk:
                    yield b
            else:
                break

# example:
for b in bytes_from_file('filename'):
    do_stuff_with(b)

有关迭代器和生成器的信息,请参阅Python文档。

对于大尺寸,我认为使用生成器不会不好,这个答案是为了读取像文件这样的东西,虽然@codeapp有一个类似的答案,我认为删除内部循环会更有意义。

def read_chunk(file_object, chunk_size=125):
    while True:
        file =  file_object.read(chunk_size)
        if not file:
            break
        yield file


#sample use 
buffer = io.BytesIO()
file = open('myfile', 'r')
for chunk in read_chunk(file):
    buffer.write(chunk)
buffer.seek(0)
// save the file or do whatever you want here

你仍然可以使用它作为一个正常的列表,我不认为这是任何用途,但是

file_list = list(read_chunk(file, chunk_size=10000))
for i in file_list:
    # do something

然后得到每个数据块的索引

for index, chunk in enumurate(read_chunk(file, chunk_size=10000)):
    #use the index as a number index
    # you can try and get the size of each chunk with this 
    length = len(chunk)

注意,要注意文件的大小,chunk_size总是以字节为单位。

要读取一个文件-一次一个字节(忽略缓冲)-你可以使用双参数iter(callable, sentinel)内置函数:

with open(filename, 'rb') as file:
    for byte in iter(lambda: file.read(1), b''):
        # Do stuff with byte

它调用file.read(1),直到没有返回b”(空字节串)。对于大文件,内存不会无限增长。你可以将buffering=0传递给open()来禁用缓冲——它保证每次迭代只读取一个字节(慢)。

With-statement自动关闭文件——包括下面的代码引发异常的情况。

尽管默认情况下存在内部缓冲,但一次处理一个字节的效率仍然很低。例如,下面是黑洞.py实用程序,它会吃掉所有给定的东西:

#!/usr/bin/env python3
"""Discard all input. `cat > /dev/null` analog."""
import sys
from functools import partial
from collections import deque

chunksize = int(sys.argv[1]) if len(sys.argv) > 1 else (1 << 15)
deque(iter(partial(sys.stdin.detach().read, chunksize), b''), maxlen=0)

例子:

$ dd if=/dev/zero bs=1M count=1000 | python3 blackhole.py

在我的机器上,当chunksize == 32768时,它处理大约1.5 GB/s,当chunksize == 1时,它只处理大约7.5 MB/s。也就是说,每次读取一个字节要慢200倍。考虑一下您是否可以重写处理以便一次使用多个字节,以及您是否需要性能。

Mmap允许您同时将文件视为bytearray和文件对象。如果需要访问两个接口,它可以作为在内存中加载整个文件的替代方案。特别是,你可以在一个内存映射文件上一次迭代一个字节,只使用简单的for循环:

from mmap import ACCESS_READ, mmap

with open(filename, 'rb', 0) as f, mmap(f.fileno(), 0, access=ACCESS_READ) as s:
    for byte in s: # length is equal to the current file size
        # Do stuff with byte

Mmap支持切片表示法。例如,mm[i:i+len]返回文件中从位置i开始的len字节。Python 3.2之前不支持上下文管理器协议;在这种情况下,需要显式调用mm.close()。使用mmap遍历每个字节比file.read(1)消耗更多的内存,但是mmap要快一个数量级。