本教程将手把手教你如何利用树莓派、Pi摄像头以及Edge Impulse平台，打造一台能够自动识别并分拣红番茄与绿番茄的智能分拣机。从硬件组装、模型训练与部署，到通过舵机控制实现分类，整个流程无需编写大量代码，非常适合AI初学者和创客实践。

一、项目概述

人工分拣番茄不仅耗时费力，而且效率低下。本教程借助机器学习实现自动化：摄像头实时拍摄番茄图像，树莓派运行经过Edge Impulse训练的图像分类模型，精准识别出红番茄或绿番茄，随后驱动两个伺服电机将番茄送入对应的收集盒中。

二、所需组件

硬件

• 树莓派（推荐Raspberry Pi 3B+或4B）
• Pi 相机模块（官方Camera Module或兼容摄像头）
• 2×伺服电机（SG90或MG90S均可）

软件

• 边缘脉冲工作室（Edge Impulse Studio）
• 树莓派OS（Raspbian Bullseye或更新版本）
• Python 3及依赖库（cv2、RPi.GPIO等）

三、Edge Impulse 入门与树莓派配置

1. 创建Edge Impulse账号与项目

访问 Edge Impulse Studio 注册并完成邮箱验证。登录后点击“Create new project”，输入项目名称（如“TomatoSorter”），并选择“Image”作为数据类型。

2. 在树莓派上安装Edge Impulse依赖

打开树莓派终端，依次执行以下命令安装Node.js、编译工具以及Edge Impulse Linux客户端：

curl -sL https://deb.nodesource.com/setup_12.x | sudo bash -
sudo apt install -y gcc g++ make build-essential nodejs sox gstreamer1.0-tools gstreamer1.0-plugins-good gstreamer1.0-plugins-base gstreamer1.0-plugins-base-apps
sudo npm install edge-impulse-linux -g --unsafe-perm

3. 运行Edge Impulse并连接摄像头

输入命令 edge-impulse-linux，按提示登录Edge Impulse账号，选择刚才创建的项目，然后选择摄像头（如果存在多个摄像头，系统会列出选项）。

注意： 请确保Pi摄像头已正确连接。参考下图连接Camera Module：

四、创建数据集

我们需要收集三类图像样本：红番茄、绿番茄以及不确定情况（比如空场景或非番茄物体）。推荐使用手机或树莓派摄像头拍摄，也可以上传已有图片。

方法一：使用手机连接Edge Impulse

1. 在Edge Impulse Studio中点击左侧“Devices”→“Connect new device”。
2. 选择“Use your phone”，系统会生成二维码。
3. 用手机上的二维码扫描工具（如Google Lens）扫描，将手机与项目绑定。

采集样本

1. 进入“Data acquisition”页面。
2. 在“Label”输入框中输入“Red Tomato”，传感器选“Camera”，点击“Start sampling”拍摄约50-60张不同角度的红番茄图片。
3. 重复上述步骤，将标签改为“Green Tomato”，再拍摄50-60张绿番茄图片。
4. 再添加一个“Uncertain”标签，拍摄约20张空背景或无关物体的图片。

小提示： 拍摄时注意光照均匀，背景简洁，番茄尽量占图像主体。训练数据与测试数据比例建议为80%/20%。

五、训练模型

1. 创建脉冲（Impulse）

点击“Create impulse”页面，然后：

1. 点击“Add processing block”，选择“Image”块，用于标准化图像并减少颜色深度。
2. 点击“Add learning block”，选择“Transfer learning (Images)”，它会加载预训练的MobileNetV2模型。
3. 点击“Sa ve impulse”。

2. 生成特征

进入“Impulse design”下的“Image”子项，点击“Generate features”选项卡，然后点击绿色的“Generate features”按钮。等待处理完成。

3. 训练模型

点击“Transfer learning”子项，保持默认设置（MobileNetV2），然后点击页面底部的“Start training”。训练可能需要几分钟，完成后会显示准确率和损失值。

本例中，训练准确率约75%，损失0.58。你可以在“Live classification”中用摄像头实时测试模型效果。

六、在树莓派上部署模型

有两种部署方式：

方法一：使用edge-impulse-linux-runner（无需编程，但无法直接控制硬件）

在树莓派终端执行：

edge-impulse-linux-runner

这会自动下载并运行最新模型，在终端显示分类结果。若需重新开始新项目，先运行：

edge-impulse-runner --clean

方法二：下载模型文件并使用Python SDK（推荐，便于控制伺服电机）

1. 下载模型文件到当前目录：
   edge-impulse-linux-runner --download modelfile.eim
2. 克隆Python SDK示例仓库：
   git clone https://github.com/edgeimpulse/linux-sdk-python
3. 进入示例文件夹，我们将使用 classify.py 脚本。在命令行运行：
   python3 classify.py modelfile.eim

运行后，终端会实时打印每个类别的置信度，如下图：

修改代码以控制伺服电机

我们需要在分类结果的基础上，根据红/绿/不确定的概率值，驱动两个舵机进行分拣。关键修改点如下（完整代码见文末）：

• 将三个类别的分数存入数组：data.append(score)
• 赋值给变量：green = round(data[0],2)，red = round(data[1],2)，uncertain = round(data[2],2)
• 设定阈值：当绿色概率≥0.45且每10帧执行一次时，舵机1转到绿色位置（PWM 12.0），舵机2转动将番茄拨入绿盒；红色同理（概率≥0.50时PWM 7.0）。

注意： 实际阈值需根据你的模型表现调整，可在测试中微调。

七、电路连接与分拣机搭建

电路图

将两个伺服电机连接到树莓派的GPIO引脚：

• 第一个舵机（用于推动番茄前进）连接到GPIO 25
• 第二个舵机（用于分拣至对应盒子）连接到GPIO 17
• 两个舵机的电源正极（红线）连接树莓派的5V引脚，负极（棕线）连接GND

小提示： 如果舵机数量较多或功耗较大，建议使用外部5V电源供电，避免烧坏树莓派。

物理结构搭建

使用2mm厚的白色Sunboard（泡沫板）制作支架和滑道。两个舵机分别安装：

• 第一个舵机：在番茄滑道入口处，通过摇臂每次将一颗番茄推入摄像头视野下方。
• 第二个舵机：在视野下方，根据分类结果将番茄拨向左侧（红盒）或右侧（绿盒）。

组装完成后参考下图：

八、测试与运行

1. 将番茄放入进料托盘。
2. 确保摄像头下方有一颗番茄。
3. 在树莓派上运行修改后的Python脚本（见下方完整代码）。
4. 观察舵机是否根据颜色正确分拣。

九、完整代码

以下代码基于官方示例修改，已集成舵机控制逻辑。请将文件保存为 tomato_sorter.py，与下载的 modelfile.eim 放在同一目录下。

#!/usr/bin/env python
import cv2
import os
import sys, getopt
import signal
import time
from edge_impulse_linux.image import ImageImpulseRunner
import RPi.GPIO as GPIO

runner = None
show_camera = False
framee_count=0

servo_pin = 25
servo1 = 17
GPIO.setmode(GPIO.BCM)
GPIO.setup(servo_pin,GPIO.OUT)
GPIO.setup(servo1, GPIO.OUT)

# 设置 PWM 进程
pwm = GPIO.PWM(servo_pin,50) # 50 Hz (20 ms PWM 周期)
pwm1 = GPIO.PWM(servo1,50)
pwm.start(7) # 旋转 90 度启动 PWM
pwm1.start(7)
pwm1.ChangeDutyCycle(2.0)
pwm.ChangeDutyCycle(2.0) #close

def now():
    return round(time.time() * 1000)

def get_webcams():
    port_ids = []
    for port in range(5):
        print("正在端口 %s 中寻找摄像头：" %port)
        camera = cv2.VideoCapture(port)
        if camera.isOpened():
            ret = camera.read()
            if ret:
                backendName =camera.getBackendName()
                w = camera.get(3)
                h = camera.get(4)
                print("在端口 %s 中找到相机 %s (%sx %s) " %(backendName,h,w, port))
                port_ids.append(port)
        camera.release()
    return port_ids

def sigint_handler(sig, frame):
    print('中断')
    if (runner):
        runner.stop()
    sys.exit(0)

signal.signal(signal.SIGINT, sigint_handler)

def help():
    print('python classify.py  <摄像头端口ID，仅当存在多于1个摄像头时才需要>')

def main(argv):
    framee_count=0
    try:
        opts, args = getopt.getopt(argv, "h", ["--help"])
    except getopt.GetoptError:
        help()
        sys.exit(2)
    for opt, arg in opts:
        if opt in ('-h','--help'):
            help()
            sys.exit()
    if len(args) == 0:
        help()
        sys.exit(2)
    model = args[0]
    dir_path = os.path.dirname(os.path.realpath(__file__))
    modelfile = os.path.join(dir_path, model)
    print('模型:' + modelfile)
    with ImageImpulseRunner(modelfile) as runner:
        try:
            model_info = runner.init()
            print('为 "' + model_info['project']['owner'] + ' / ' + model_info['project']['name'] + '" 加载了runner')
            labels = model_info['model_parameters']['labels']
            if len(args)>= 2:
                videoCaptureDeviceId = int(args[1])
            else:
                port_ids = get_webcams()
                if len(port_ids) == 0:
                    raise Exception('找不到任何网络摄像头')
                if len(args) <= 1 and len(port_ids) > 1:
                    raise Exception("找到多个摄像头。将摄像头端口 ID 作为第二个参数添加到此脚本中")
                videoCaptureDeviceId = int(port_ids[0])
            camera = cv2.VideoCapture(videoCaptureDeviceId)
            ret = camera.read()[0]
            if ret:
                backendName = camera.getBackendName()
                w = camera.get(3)
                h = camera.get(4)
                print("已选择端口 %s 中的摄像机 %s (%sx %s)。" %(backendName,h,w, videoCaptureDeviceId))
            camera.release()
            else:
                raise Exception("无法初始化选定的相机。")
            next_frame = 0 # 此处限制为 ~10 fps
            for res, img in runner.classifier(videoCaptureDeviceId):
                if (next_frame > now()):
                    time.sleep((next_frame - now()) / 1000)
                # print('分类运行响应', res)
                data = []
                framee_count = framee_count +1
                print("帧数：", framee_count)
                if "classification" in res["result"].keys():
                    print('结果 (%d ms.) ' % (res['timing']['dsp'] + res['timing']['classification']), end='')
                    for label in labels:
                        score = res['result']['classification'][label]
                        # print(score)
                        print('%s: %.2f\t' % (label, score), end='')
                        data.append(score)
                    print('', flush=True)
                    green=round(data[0],2)
                    red = round(data [1],2)
                    uncertain=round(data[2],2)
                    print(green, red, uncertain)
                    if (green >=0.25 and framee_count%10 ==0):
                        while(green >= 0.25):
                            pwm1.ChangeDutyCycle(12.0)
                            print("检测到绿色番茄")
                            time.sleep(0.500)
                            pwm1.ChangeDutyCycle(2.0) #close
                            time.sleep(0.250)
                            pwm.ChangeDutyCycle(7.0)
                            time.sleep(0.450)
                            pwm.ChangeDutyCycle(2.0)
                            green=0.01
                        # time.sleep(2)
                    if (red >=0.50 and framee_count%10 ==0):
                        while(red >= 0.50):
                            pwm1.ChangeDutyCycle(7.0)
                            print("检测到红番茄")
                            time.sleep(0.500)
                            pwm1.ChangeDutyCycle(2.0)
                            time.sleep(0.250)
                            pwm.ChangeDutyCycle(7.0)
                            time.sleep(0.450)
                            pwm.ChangeDutyCycle(2.0)
                            red=0.01
                        # time.sleep(2)
                    else:
                        time.sleep(0.01)
                # print('%s: %.2f\t' % (Green,Red,Uncertain), end ='')
                if (show_camera):
                    cv2.imshow('edgeimpulse', img)
                    if cv2.waitKey(1) == ord('q'):
                        break
            elif "bounding_boxes" in res["result"].keys():
                print('找到 %d 个边界框 (%d ms.)' % (len(res["result"]["bounding_boxes"]), res['timing']['dsp'] + res['timing']['classification']))
                for bb in res["result"]["bounding_boxes"]:
                    print('\t%s (%.2f): x=%d y=%d w=%d h=%d' % (bb['label'], bb['value'], bb['x'], bb['y'], bb['width'], bb['height']))
            next_frame = now() + 100
        finally:
            if (runner):
                runner.stop()
        # framee_count=0
if __name__ == "__main__":
    main(sys.argv[1:])
cap.release()
cv2.destroyAllWindows()

十、常见问题与解答

Q1：训练准确率只有75%，该如何提高？

A： 可以尝试以下方法：
- 增加训练样本数量（每个类别至少100张）
- 确保样本光线、角度多样化
- 调整迁移学习的训练参数（如学习率、训练周期）
- 使用更深的模型（如MobileNetV2 1.0），但推理时间会增加。

Q2：运行时报错“没有找到摄像头”怎么办？

A： 检查摄像头是否物理连接牢固。在终端运行 ls /dev/video* 查看是否有设备节点。若使用树莓派官方摄像头，需在 raspi-config 中启用Camera接口。

Q3：舵机不动作或动作异常？

A： 首先检查舵机接线：信号线是否正确连接到GPIO，电源和地线是否接好。其次确认PWM频率为50Hz，占空比范围（2.0~12.0对应0°~180°）。可在Python中单独测试舵机转动。

Q4：如何将模型部署到其他设备（如jetson nano）？

A： Edge Impulse支持导出为TensorFlow Lite、ONNX等格式，可在其他Linux设备上使用对应的SDK运行。流程类似，只需安装对应设备上的Edge Impulse运行时或使用Python SDK。

十一、总结

通过本教程，你已学会从零开始搭建一台基于AI视觉的番茄分拣机：从数据集采集、模型训练到树莓派部署，并通过舵机实现物理分拣。项目不仅可扩展至其他果蔬分类（如苹果、柠檬），还能根据腐烂状态进行检测。欢迎在评论区或论坛中交流改进方案。