如何在Python中将类Bytes对象转换为uint16数组？

需求背景 #

需要从UDP流中接收二进制数据，将其转换为uint16类型的N×16数组，对应Matlab中的如下实现：

Matlab实现代码 #

1. 接收UDP并转为uint16列向量

% Build UDP Connection and collect binary data
udpr = dsp.UDPReceiver('LocalIPPort',5005,'ReceiveBufferSize',3200,'MessageDataType','uint16','MaximumMessageLength',640)
setup(udpr)
data = udpr();

2. 将列向量转为N×16数组

rows_result = length(data)/16;
result = zeros(rows_result,16);
pointer=0;
for i=1:rows_result
    for j=1:16
        result(i,j)=data(pointer+j);
    end
    pointer=pointer+16;
end

Python实现问题 #

尝试用Python接收UDP字节数据时，无法正确将2字节的uint16数据解析出来，list()只能读取单字节，struct.unpack、int.from_bytes和array模块的尝试均未成功。以下是测试用的初始代码：

原测试代码 #

发送端（sender.py）

import socket
import time
import struct
import numpy as np

np.random.seed(1)

UDP_IP = "127.0.0.1"
UDP_PORT = 5005

print("UDP target IP: %s" % UDP_IP)
print("UDP target port: %s" % UDP_PORT)

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

while True:
    # 生成20×16的uint16随机数组
    input_matrix = np.random.randint(0, 4095, [20,16],dtype=np.uint16)
    print(input_matrix)
    
    MESSAGE = input_matrix
    sock.sendto(MESSAGE, (UDP_IP, UDP_PORT))
    time.sleep(3)

接收端（receiver.py）

import socket

UDP_IP = "127.0.0.1"
UDP_PORT = 5005

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind((UDP_IP, UDP_PORT))

while True:
    data, addr = sock.recvfrom(1024)
    print(data)

解决方案 #

1. 修正发送端：正确序列化numpy数组 #

直接传递numpy数组无法被UDP正确发送，需要将数组转换为字节流，同时保证字节序与Matlab匹配：

import socket
import time
import numpy as np

np.random.seed(1)

UDP_IP = "127.0.0.1"
UDP_PORT = 5005

print("UDP target IP: %s" % UDP_IP)
print("UDP target port: %s" % UDP_PORT)

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

while True:
    # 生成N×16的uint16随机数组（此处N=20）
    input_matrix = np.random.randint(0, 4095, [20,16], dtype=np.uint16)
    print(input_matrix)
    
    # 将numpy数组转为字节流，使用C风格内存布局（与Matlab默认逻辑一致）
    MESSAGE = input_matrix.tobytes('C')
    sock.sendto(MESSAGE, (UDP_IP, UDP_PORT))
    time.sleep(3)

2. 接收端：解析字节流并转为N×16数组 #

使用numpy直接解析字节数据，无需手动循环，高效匹配Matlab逻辑：

import socket
import numpy as np

UDP_IP = "127.0.0.1"
UDP_PORT = 5005

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind((UDP_IP, UDP_PORT))

while True:
    data, addr = sock.recvfrom(1024)
    # 将字节数据解析为uint16数组，字节序与发送端保持一致
    uint16_array = np.frombuffer(data, dtype=np.uint16)
    # 自动计算行数并重塑为N×16的二维数组
    result = uint16_array.reshape(-1, 16)
    print(result)

关键说明 #

• 字节序一致性：若Matlab使用特定字节序（如大端），可在numpy中指定dtype='>u2'（大端uint16）或'<u2'（小端uint16）来匹配。
• 自动行数计算：reshape(-1,16)会根据总元素数自动推导行数N，等价于Matlab中length(data)/16的逻辑。
• 缓冲区适配：确保recvfrom的缓冲区大小足够容纳单次发送的所有字节（例如20×16的uint16数组共640字节，1024字节缓冲区完全足够）。

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