Welcome to DOFBOT-Pi repository

Voice Control Yolov11 Garbage Recognition Sorting

1. Function Description

After the program runs, wake up the voice module through "Hello yahboom", then give the voice command "Start garbage sorting". The robotic arm will recognize the garbage label wooden blocks on the desktop, then grasp the recognized garbage label wooden blocks, announce the garbage name and which type it belongs to, and finally place them at the set position according to the garbage type. After placement is complete, the voice will announce "Placement complete".

2. Function Startup

After entering the docker container, enter the following commands in the docker terminal:

# Start USB camera node and inverse kinematics node
ros2 launch dofbot_info camera_arm_kin.launch.py
# Start speech recognition and broadcast node
ros2 launch yahboom_speech speech.launch.py
# Start garbage recognition voice broadcast node
ros2 run dofbot_voice_ctrl yolov11_garbage
# Start grasping node
ros2 run dofbot_sorting grasp

After the program runs, after waking up with "Hello yahboom", give the command "Start garbage sorting". The program will recognize the garbage label wooden blocks on the desktop, and the voice module will announce the garbage name and which category of garbage it belongs to, then grasp the garbage label wooden blocks, place them at the set position, and finally announce placement complete.

3. Code Analysis

3.1 yolov11_garbage

After entering the docker container, the source code path is located at:

xxxxxxxxxx
/root/dofbot_ws/src/dofbot_voice_ctrl/dofbot_voice_ctrl/Yolov11/yolov11_garbage.py

Import necessary libraries:

xxxxxxxxxx
import rclpy
from rclpy.node import Node
import cv2 
import numpy as np
from sensor_msgs.msg import Image
from cv_bridge import CvBridge
import cv2 as cv
from dofbot_interface.srv import Kinemarics
from dofbot_interface.msg import *
from std_msgs.msg import Bool,Int16,Int8
import time
from Arm_Lib import Arm_Device
from ultralytics import YOLO

Load garbage recognition model:

xxxxxxxxxx
model = YOLO("/root/ultralytics/ultralytics/best.onnx")

Define garbage types:

xxxxxxxxxx
self.recyclable_waste=['Newspaper','Zip_top_can','Book','Old_school_bag']
self.toxic_waste=['Syringe','Expired_cosmetics','Used_batteries','Expired_tablets']
self.wet_waste=['Fish_bone','Egg_shell','Apple_core','Watermelon_rind']
self.dry_waste=['Toilet_paper','Peach_pit','Cigarette_butts','Disposable_chopsticks']

Create publishers and subscribers:

xxxxxxxxxx
# Create publisher to publish calculated servo 5 angle reference value topic messages
self.TargetJoint5_pub = self.create_publisher(Int16, "set_joint5", 10)
# Create publisher to publish recognized position information topic messages
self.pos_info_pub = self.create_publisher(AprilTagInfo, "PosInfo", qos_profile=10)
# Create subscriber to subscribe to grasp placement completion topic messages
self.subscription = self.create_subscription(Bool,'grasp_done',self.GraspStatusCallback,qos_profile=1)
# Create subscriber to subscribe to image topic
self.subscription = self.create_subscription(Image,'/image_raw',self.ImageCallback,1)
# Create subscriber to subscribe to speech recognition result topic
self.sub_voice = self.create_subscription(Int8,"voice_result",self.getVoiceResultCallBack,1)
# Create publisher to publish voice broadcast content topic messages
self.pub_playID = self.create_publisher(Int8, "player_id", qos_profile=10)

Grasp placement completion topic callback function GraspStatusCallback:

xxxxxxxxxx
def GraspStatusCallback(self,msg):
    print("**")
    # If the received topic is True, it means the grasp placement action is completed
    if msg.data == True:
        # Create voice broadcast content topic message object
        play_id = Int8()
        # Assign data to voice broadcast content topic message, broadcast audio sequence number 81, "Placement complete"
        play_id.data = 81
        # Publish voice broadcast content topic message
        self.pub_playID.publish(play_id)
        self.pubPos_flag = True

Speech recognition result topic callback function getVoiceResultCallBack:

xxxxxxxxxx
# Change self.pubPos_flag value to True based on speech recognition result, indicating that current position information topic can be published
def getVoiceResultCallBack(self,msg):
    if msg.data == 108:
        self.pubPos_flag = True

Image topic processing function ImageCallback:

xxxxxxxxxx
def ImageCallback(self,color_frame):
    # Process received image topic messages, convert to image
    rgb_image = self.rgb_bridge.imgmsg_to_cv2(color_frame,'rgb8')
    result_image = np.copy(rgb_image)
    frame = np.copy(rgb_image)
    # Call yolov11 model, pass image for recognition, store recognition results in results
    results = model(frame)
    key = cv2.waitKey(10)
    # If spacebar is pressed, it means garbage label position information topic can be published
    if key == 32:
        self.pubPos_flag = True 
    # Extract recognition results
    boxes = results[0].boxes
    # If recognition results are not empty, it means garbage on the garbage label is recognized
    if boxes != [None] and self.start_flag == True:
        for box in boxes:  # detections per image
            # Extract the top-left and bottom-right recognition boxes of the garbage label
            x_min, y_min, x_max, y_max = map(int, box.xyxy[0])
            class_id = int(box.cls)
            confidence = float(box.conf)
            label = f"{model.names[class_id]} {confidence:.2f}"
            print("model.names[class_id]: ",model.names[class_id])
            # Calculate center point coordinates
            center_x = (x_min + x_max) // 2
            center_y = (y_min + y_max) // 2
            # Based on center point coordinates, convert coordinates to normalized coefficients a and b, x-axis offset coefficient a, y-axis offset coefficient b, the calculated ba value is the center point coordinates in world coordinate system, this value needs to be sent to the sorting grasp node.
            (a, b) = (round(((320 - center_x) / 4000), 5), round(((480 - center_y) / 3000) * 0.8+0.15, 5))
            if self.pubPos_flag == True:
                # self.pubPos_flag value is False, indicating prohibition of sending position topic, preventing sending recognized position messages again during grasping process
                self.pubPos_flag = False
                # Create position information topic message object
                center = AprilTagInfo()
                center.x = b
                center.y = a
                center.z = 0.03
                # Get garbage name on the garbage label
                self.name = str(model.names[class_id])
                # Determine which category the garbage belongs to, then publish voice broadcast content topic message based on garbage name and assign value to center.id based on garbage type
                if str(model.names[class_id]) in self.recyclable_waste:
                    center.id = 1
                    if self.name == "Newspaper":
                        self.play_id.data = 96
                        self.pub_playID.publish(self.play_id)     
                    elif self.name == "Zip_top_can":                
                        self.play_id.data = 94
                        self.pub_playID.publish(self.play_id)
                    elif self.name == "Book":                
                        self.play_id.data = 97
                        self.pub_playID.publish(self.play_id)
                    elif self.name == "Old_school_bag":
                        self.play_id.data = 95
                        self.pub_playID.publish(self.play_id) 
                elif str(model.names[class_id]) in self.toxic_waste:
                    center.id = 2
                    if self.name == "Syringe":
                        self.play_id.data = 98
                        self.pub_playID.publish(self.play_id)
                    elif self.name == "Expired_cosmetics":
                        self.play_id.data = 100
                        self.pub_playID.publish(self.play_id)
                    elif self.name == "Expired_tablets":
                        self.play_id.data = 101
                        self.pub_playID.publish(self.play_id)
                    elif self.name == "Used_batteries":
                        self.play_id.data = 99
                        self.pub_playID.publish(self.play_id)                            
                        
                elif str(model.names[class_id]) in self.wet_waste:
                    center.id = 3
                    if self.name == "Fish_bone":
                        self.play_id.data = 102
                        self.pub_playID.publish(self.play_id)
                    
                elif str(model.names[class_id]) in self.dry_waste:
                    center.id = 4
                    if self.name == "Toilet_paper":
                        self.play_id.data = 109
                        self.pub_playID.publish(self.play_id)     
                    elif self.name == "Disposable_chopsticks":
                        self.play_id.data = 106
                        self.pub_playID.publish(self.play_id)
                    elif self.name == "Cigarette_butts":
                        self.play_id.data = 107
                        self.pub_playID.publish(self.play_id)
                    elif self.name == "Peach_pit":
                        self.play_id.data = 108
                        self.pub_playID.publish(self.play_id)                        
                print("center: ",center)
                # Publish position information topic of currently recognized garbage label
                self.pos_info_pub.publish(center)

4. Node Communication Graph

After entering the docker container, enter the following command in the docker terminal:

xxxxxxxxxx
ros2 run rqt_graph rqt_graph