高效转换大规模统一字节序列至NumPy数组：np.frombuffer实战指南

import numpy as np

# 示例数据：实际数据量可能高达千万级别
# 每个元组包含3个450字节的序列
data = [
    (
        b'

	
' * 45, # 450 bytes series
        b'
' * 45, # also 450 bytes
        b'	
' * 45,
    ),
    (
        b'
	' * 45, # 450 bytes series
        b'

' * 45, # also 450 bytes
        b'' * 45,
    ),
]
# 假设实际数据有 N 个这样的元组，例如 N=2
N_tuples = len(data) # 2
N_series_per_tuple = len(data[0]) # 3
bytes_length = len(data[0][0]) # 450

# 将列表转换为NumPy数组，指定dtype为np.bytes_
# 由于所有字节序列长度一致，NumPy会推断为 '|S450' 这样的固定长度字节字符串类型
bytes_array = np.array(data, dtype=np.bytes_)

print(f"原始字节数组形状: {bytes_array.shape}") # (2, 3)
print(f"原始字节数组数据类型: {bytes_array.dtype}") # |S450 (或类似)
print(f"原始字节数组内容示例:
{bytes_array[0, 0][:20]}...") # 打印部分内容

# 将bytes_array扁平化，然后通过frombuffer解释为uint8
# bytes_array_flat = bytes_array.reshape(-1) # 这一步是可选的，但有助于理解为单一连续内存
# 实际上，直接对bytes_array使用frombuffer即可，因为它内部是连续的
uint8_flat_array = np.frombuffer(bytes_array, dtype=np.uint8)

print(f"
解释为uint8后的扁平数组形状: {uint8_flat_array.shape}")
print(f"解释为uint8后的扁平数组数据类型: {uint8_flat_array.dtype}") # uint8
print(f"解释为uint8后的扁平数组内容示例:
{uint8_flat_array[:20]}")

# 重塑为目标形状 (N_tuples, N_series_per_tuple, bytes_length)
final_uint8_array = uint8_flat_array.reshape(N_tuples, N_series_per_tuple, bytes_length)

print(f"
最终uint8数组形状: {final_uint8_array.shape}") # (2, 3, 450)
print(f"最终uint8数组数据类型: {final_uint8_array.dtype}") # uint8
print(f"最终uint8数组内容示例 (第一个元组的第一个序列):
{final_uint8_array[0, 0, :20]}")

import numpy as np

# 假设实际数据有 N 个这样的元组，每个元组包含 M 个 K 字节的序列
# 示例数据：N=2, M=3, K=10 (为了演示方便，实际问题中 K=450)
N_tuples = 2
N_series_per_tuple = 3
bytes_length = 10 # 实际问题中是 450

data = [
    (
        b'

	

', # 10 bytes series
        b'
', # also 10 bytes
        b'	
',
    ),
    (
        b'
	
', # 10 bytes series
        b'

', # also 10 bytes
        b'',
    ),
]

print("--- 原始数据结构示例 ---")
print(f"数据总条目数 (N): {N_tuples}")
print(f"每个条目中的字节序列数 (M): {N_series_per_tuple}")
print(f"每个字节序列的长度 (K): {bytes_length}")
print(f"第一个数据元组:
{data[0]}
")

# 第一步：将列表转换为NumPy数组，dtype='S<length>'确保内存连续
# np.bytes_ 会根据实际字节长度自动推断为 |S<length>
bytes_array = np.array(data, dtype=np.bytes_)

print("--- 第一步结果：固定长度字节字符串数组 ---")
print(f"NumPy数组形状: {bytes_array.shape}") # (N, M)
print(f"NumPy数组数据类型: {bytes_array.dtype}") # |S<K> (e.g., |S10)
print(f"数组内容示例 (第一个元素):
{bytes_array[0]}
")

# 第二步：使用np.frombuffer将数组的原始内存解释为uint8数据
# bytes_array本身支持缓冲区协议，可以直接作为frombuffer的输入
uint8_flat_array = np.frombuffer(bytes_array, dtype=np.uint8)

print("--- 第二步结果：扁平化的uint8数组 ---")
print(f"扁平数组形状: {uint8_flat_array.shape}") # (N * M * K,)
print(f"扁平数组数据类型: {uint8_flat_array.dtype}") # uint8
print(f"扁平数组内容示例 (前20个字节):
{uint8_flat_array[:20]}
")

# 第三步：重塑数组以匹配预期维度
# 目标形状为 (N_tuples, N_series_per_tuple, bytes_length)
final_uint8_array = uint8_flat_array.reshape(N_tuples, N_series_per_tuple, bytes_length)

print("--- 第三步结果：最终的uint8多维数组 ---")
print(f"最终数组形状: {final_uint8_array.shape}") # (N, M, K)
print(f"最终数组数据类型: {final_uint8_array.dtype}") # uint8
print(f"最终数组内容示例 (第一个元组的第一个序列):
{final_uint8_array[0, 0, :]}")
print(f"验证：原始字节 `b'\n\x0f\n\t\x0c\x00\x00\x01\x07\x06'` 对应 `[10, 15, 10, 9, 12, 0, 0, 1, 7, 6]`")