You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

socket传输base64解码图像报expected str not numpy.ndarray错误如何解决

问题原因
  • open() 函数的第一个入参要求是文件路径类对象,你传入的img是cv2.imdecode返回的numpy矩阵,类型不匹配触发报错
  • 你的逻辑存在冗余:base64解码后的decoded_data本身就是图像文件的完整二进制内容,不需要额外通过opencv转码再读取文件
修复后的客户端代码

如果你不需要对图像做处理,可以直接发送解码后的二进制内容,省略opencv转码步骤:

import base64
import socket

def main(data):
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.connect(('192.168.0.14', 9999))
    BUFFER_SIZE = 4096*4

    # base64解码后直接就是图像二进制,不需要额外转码
    decoded_data = base64.b64decode(data)
    
    # 直接分块发送二进制内容即可
    start = 0
    data_len = len(decoded_data)
    while start < data_len:
        end = min(start + BUFFER_SIZE, data_len)
        client.send(decoded_data[start:end])
        start = end
    client.close()

如果你确实需要对图像做裁剪、改格式等处理,处理完成后再编码为二进制发送即可:

import numpy as np
import cv2
import base64
import socket

def main(data):
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.connect(('192.168.0.14', 9999))
    BUFFER_SIZE = 4096*4

    decoded_data = base64.b64decode(data)
    # fromstring已弃用,替换为frombuffer兼容性更好
    np_data = np.frombuffer(decoded_data, np.uint8)
    img = cv2.imdecode(np_data, cv2.IMREAD_UNCHANGED)
    
    # 这里可以添加你的图像处理逻辑,比如缩放、改色等
    # img = cv2.resize(img, (640, 480))
    
    # 处理完成后编码回jpeg格式二进制
    ret, encoded_img = cv2.imencode('.jpeg', img)
    img_bytes = encoded_img.tobytes()
    
    # 分块发送
    start = 0
    data_len = len(img_bytes)
    while start < data_len:
        end = min(start + BUFFER_SIZE, data_len)
        client.send(img_bytes[start:end])
        start = end
    client.close()
服务端代码优化

原有服务端的时间戳在启动时生成,会导致所有接收的图像使用同一个文件名,后续文件会覆盖之前的内容,优化后代码如下:

import socket
import time
import os

# 提前创建保存目录避免报错
os.makedirs('frames_saved', exist_ok=True)

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('192.168.0.14', 9999))
server.listen()

BUFFER_SIZE = 4096*4

while True:
    client_socket, _ = server.accept()
    # 每次接收新连接时生成新的时间戳作为文件名,避免覆盖
    date_string = time.strftime("%Y-%m-%d-%H-%M-%S")
    save_path = os.path.join('frames_saved', f'{date_string}.jpeg')
    with open(save_path, 'wb') as file:
        recv_data = client_socket.recv(BUFFER_SIZE)
        while recv_data:
            file.write(recv_data)
            recv_data = client_socket.recv(BUFFER_SIZE)
    client_socket.close()

内容的提问来源于stack exchange,提问作者user11597888

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.09.25 11:24:03