多台树莓派本地组网通信及图片文本传输实现方法咨询

前期准备 #

• 所有树莓派接入同一局域网（WiFi/有线交换机均可），通过ip a命令查询各设备局域网IP，给主机在路由器后台绑定静态IP，避免IP变动导致连接失败
• 确认所有设备已安装Python3环境，树莓派官方系统默认自带，缺失依赖可通过pip3 install 依赖包名安装

推荐实现方案 #

方案1：Flask HTTP传输（新手最易上手） #

无需处理复杂的网络底层逻辑，主机运行轻量化Web服务开放上传接口，从机通过POST请求传输文件，文本、图片格式都通用。

主机端代码

from flask import Flask, request
import os

app = Flask(__name__)
SAVE_PATH = "./received_files"
os.makedirs(SAVE_PATH, exist_ok=True)

@app.route("/upload", methods=["POST"])
def upload():
    if "file" not in request.files:
        return "无有效文件", 400
    file = request.files["file"]
    if file.filename == "":
        return "文件名为空", 400
    file.save(os.path.join(SAVE_PATH, file.filename))
    return f"文件{file.filename}上传成功", 200

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=False)

从机端代码

import requests

# 替换为你主机的实际局域网IP
HOST_IP = "192.168.xxx.xxx"
UPLOAD_API = f"http://{HOST_IP}:5000/upload"

def send_file(file_path):
    with open(file_path, "rb") as f:
        resp = requests.post(UPLOAD_API, files={"file": f})
        print(resp.text)

if __name__ == "__main__":
    # 替换为你要传输的文件路径
    send_file("./test.png")

使用步骤

• 主机执行pip3 install flask安装依赖，运行python3 主机代码文件名.py启动服务
• 从机执行pip3 install requests安装依赖，修改代码内的主机IP后，运行python3 从机代码文件名.py即可传输文件
• 注意：连接失败时先排查主机是否开放5000端口，树莓派默认未开启防火墙的情况下无需额外配置

方案2：原生Socket TCP传输（无第三方依赖） #

不需要安装额外依赖，适合需要自定义传输逻辑的场景，稳定性更高。

主机端代码

import socket
import os

SAVE_PATH = "./received_files"
os.makedirs(SAVE_PATH, exist_ok=True)
HOST = "0.0.0.0"
PORT = 6666
BUFFER_SIZE = 4096

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind((HOST, PORT))
server.listen(5)
print(f"主机已启动，监听端口{PORT}")

while True:
    conn, addr = server.accept()
    # 先接收文件名长度、文件名、文件大小
    filename_len = int.from_bytes(conn.recv(4), byteorder="big")
    filename = conn.recv(filename_len).decode("utf-8")
    file_size = int.from_bytes(conn.recv(8), byteorder="big")
    # 接收文件内容
    received = 0
    with open(os.path.join(SAVE_PATH, filename), "wb") as f:
        while received < file_size:
            data = conn.recv(BUFFER_SIZE)
            if not data:
                break
            f.write(data)
            received += len(data)
    print(f"文件{filename}接收完成")
    conn.close()

从机端代码

import socket
import os

# 替换为你主机的实际局域网IP
HOST_IP = "192.168.xxx.xxx"
PORT = 6666
BUFFER_SIZE = 4096

def send_file(file_path):
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.connect((HOST_IP, PORT))
    filename = os.path.basename(file_path)
    file_size = os.path.getsize(file_path)
    # 先发文件元信息
    client.send(len(filename).to_bytes(4, byteorder="big"))
    client.send(filename.encode("utf-8"))
    client.send(file_size.to_bytes(8, byteorder="big"))
    # 发文件内容
    with open(file_path, "rb") as f:
        while True:
            data = f.read(BUFFER_SIZE)
            if not data:
                break
            client.sendall(data)
    client.close()
    print("文件发送完成")

if __name__ == "__main__":
    # 替换为你要传输的文件路径
    send_file("./test.txt")

使用步骤

• 主机直接运行python3 主机代码文件名.py启动服务
• 从机修改代码内的主机IP后，运行python3 从机代码文件名.py即可传输文件

问题排查方向 #

• 先执行ping 主机IP验证从机和主机的网络连通性，ping不通优先排查局域网配置
• 端口被占用时，更换代码内的端口号即可，建议选择1024~65535之间未被使用的端口
• 大文件传输可后续自行添加断点续传、MD5校验逻辑，前期先传输小文件验证通路正常

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