WebSocket传输cv2.VideoCapture图像解码失败求助

我帮你排查了代码里的问题，核心出在base64编码后的字符串处理和几个细节上，下面一步步给你修复：

问题根源 #

1. 客户端里你直接把base64.b64encode返回的字节串转成字符串时，用了str(encoded_image)，这会把字节串变成类似"b'abcdef...'"的格式，带了多余的b'前缀和'后缀，服务端解码时自然会失败。
2. 服务端接收的shape是JSON序列化后的列表，而np.reshape需要元组类型。
3. np.fromstring已经被官方弃用，推荐使用np.frombuffer来处理字节数据。

修复后的客户端代码 #

ret, image_np = cap.read()
if not ret:
    print("Failed to capture image")
    # 可根据需求添加错误处理，比如重试或退出
IMAGE_SHAPE = image_np.shape
# 先把numpy数组转成原始字节流，再做base64编码并转成纯字符串
encoded_image = base64.b64encode(image_np.tobytes()).decode('utf-8')
print(type(encoded_image))
payload = {
    'from': 'rasp',
    'image': encoded_image,
    'shape': IMAGE_SHAPE,
}
data = json.dumps(payload)
try:
    # Send encoded image data.
    await websocket.send(data)
    # Receive server message
    received_data = await websocket.recv()
    print('< {}'.format(received_data))
    
    # 如果需要在客户端显示服务端返回的图像，可参考以下代码
    # received_payload = json.loads(received_data)
    # decoded_image = base64.b64decode(received_payload['image'])
    # np_image = np.frombuffer(decoded_image, dtype=np.uint8)
    # source = np_image.reshape(tuple(received_payload['shape']))
    # cv2.imshow('Received Image', source)
    # cv2.waitKey(1)
    
    return websocket
except Exception as e:
    print(f'WebSocket send or receive error: {e}')
    exit(1)

修复后的服务端代码 #

async def server_on(websocket, path):
    response_payload = {
        'from': 'image-server',
        # 后续可添加人像检测等处理结果
        # 'result': {
        #     'isPerson': False,
        #     'centerPoint': (0, 0),
        # },
    }
    try:
        payload = await websocket.recv()
        receive_data = json.loads(payload)
        
        # 修复解码流程
        decoded_image = base64.b64decode(receive_data['image'])
        # 使用np.frombuffer替代弃用的fromstring，同时将shape转成元组
        image_np = np.frombuffer(decoded_image, dtype=np.uint8)
        source = image_np.reshape(tuple(receive_data['shape']))
        
        # 现在可以正常用cv2显示图像了
        cv2.imshow('Received from Client', source)
        cv2.waitKey(1)
        
        # 发送响应给客户端
        await websocket.send(json.dumps(response_payload))
    except Exception as e:
        print(f'Server error: {e}')
        await websocket.close()
        return

关键修改点说明 #

• 客户端编码优化：先通过image_np.tobytes()把numpy数组转成原始字节，再做base64编码，最后用decode('utf-8')转成纯字符串，彻底避免了多余的b'前缀。
• 服务端shape处理：用tuple(receive_data['shape'])把JSON序列化后的列表转成numpy需要的元组类型。
• 字节转numpy数组：替换弃用的np.fromstring为np.frombuffer，符合numpy最新规范。
• 错误信息增强：添加了异常详情打印，方便快速定位问题。

这样修改后，服务端就能正确解码图像，并用cv2.imshow()正常显示了。

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