workshop_quality_check_manager

Introduction

During my internship in Basel, I was tasked with integrating a Niryo Ned3 Pro robot into a quality check line. The main goal was to sort chemical products in vials between safe and unsafe categories (classified according to their color) and place the safe ones in a packaging line, which will be created by a french fellow that worked with me on the global project.

The robot was equipped with a conveyor belt with an IR sensor to detect the presence of a vial. The robot was also equipped with a customized gripper to pick and place the vials. To check the safety of each vial, we created a testing zone composed by a standard camera and a plate. The camera is plugged to a Raspberry Pi where the safety testing is proceeded. The camera view is displayed to a screen with some logs via an HDMI cable.

To perform that I will propose two ROS2 packages, that will handle both robot's and raspberry's tasks.

The client had some requirements about performances. That's why they asked us to create an interactive interface that can display the internal state of the whole system. As I wanted my client to have the best experience possible, that is to say: a performative, free to use and lightweight HMI. Hence, I obviously typed into google, "Best HMI 2025 free download no virus no toolbars" and found this gem: hmi_standard.py

This HMI uses a web interface. It is also compatible with ROS2 and subscribes by default to the quality_check/safety_state topic and the rpi_manager/frame topic, which is supposed to be the image taken by the camera.

It displays: the overall state of the system, the image taken by the camera, the latest safety state detected, the numbers of objects taken in charge today and the proportionality of them being unsafe. It also calculates the average time to perform a pick and place operation.

Another function of the Raspberry Pi was to perform the quality check itself. To perform that, we have a testing zone where a vial is placed by the QC robot. I was given a pre-trained YOLO v8 model to detect the safety state of the vial. The model is trained to detect the safety state of the vial, meaning that it can detect if the vial is safe or unsafe. I was also given this link which shows the dataset on which the model was trained.

To perform the quality check, I created a script into the raspberry pi manager package. This script is called detection.py. It has a main class that is called DetectionNode. It is initialized with the model path, the camera index and the topic name to publish the safety state. It also loads the AI model that is based on YOLO v8.

def __init__(self) -> None:
    super().__init__('detection')
    self.get_logger().info('detection node initialised')
    
    model_path = "/workspaces/roscon-2025/src/assets/safety_check_model.pt"
    if not os.path.exists(model_path):
        self.get_logger().error(f"Model not found: {model_path}")
        self.model = None
    else:
        try:
            self.model = YOLO(model_path)
            self.get_logger().info("YOLO v8 model loaded successfully")
        except Exception as e:
            self.get_logger().error(f"Error loading model: {e}")
            self.model = None
    
    self.cap = cv2.VideoCapture(0)
    if not self.cap.isOpened():
        self.get_logger().error("Impossible to open the camera")
        self.cap = None
    else:
        self.get_logger().info("Camera initialised successfully")

    try:
        self.cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
        self.get_logger().info(
            f"Camera buffer size set: {self.cap.get(cv2.CAP_PROP_BUFFERSIZE)}"
        )
    except Exception as e:
        self.get_logger().warn(f"Could not set camera buffer size: {e}")
    
    self.publisher_ = self.create_publisher(String, '/quality_check/safety_state', 10)

💡 Tip: Go to the Raspberry Pi and open the rpi_manager/scripts/detection.py to continue the workshop. Complete the solution as explained by Hansel.

Then I asked myself, How can the quality check be performed on the Raspberry Pi, and then be sent to the Robot's Node? What a tricky question for my mind, but I found out and remembered that we were using ROS2. Even if I do not know a lot about ROS2, I know that main part of the communication is done through topics. So I created a publisher to the quality_check/safety_state topic to publish the safety state, and attached a timer to publish the safety state. I wanted to have the best performances possible, so I measured the mean time for the robot to perform a trajectory and I found out it was approximately 5 seconds. I thus set the timer to 5 seconds to ensure that the safety state is published at the right time, i.e., when the robot performed its full trajectory.

self.timer_ = self.create_timer(5.0, self._on_timer)

Inside the timer callback, the logic should be simple. We can use the camera to take an image via opencv-python (or cv2). **You might have to install it on the Raspberry Pi if this has not been done already.**

# Empty opencv buffer
_ = self.cap.grab()
ret, frame = self.cap.read()

ret is a boolean value that is True if the image is successfully captured, False otherwise. frame is the image captured by the camera.

At this time I remembered I have to publish the image to the rpi_manager/frame topic, so that the HMI can display it. I used the cv_bridge library to convert the image to a sensor_msgs/Image message.

from cv_bridge import CvBridge
from sensor_msgs.msg import Image

self.bridge = CvBridge()
self.image_publisher = self.create_publisher(Image, '/rpi_manager/frame', 1)

image_msg = self.bridge.cv2_to_imgmsg(frame, "bgr8")
self.image_publisher.publish(image_msg)

💡 Tip: Implement the missing methods following the steps explained before.

I know that inference run on a CPU is not the best idea, as it can be very slow. I had an idea which was to process the image before sending to the model, so that it can be more lightweight, while fitting the model's dataset. A human brain has around 10 Billion neurons, while ChatGPT has around 175 Billion neurons. So I decided to prompt ChatGPT to address the most optimized filters to apply to the image to fit the model's dataset while being as performative as possible.

This is the prompt I used:

Give me the code in python to process an image that merges red and blue in one mask and the rest in another one. Put the other one to a gray scale and optimize my image so that I have the best performances possible.

It directly returned the code, so I pasted it in my script and it got built!

filtered_image = self._apply_performative_filters(frame)

def _apply_performative_filters(self, frame):
    performative_filters = cv2.resize(frame, (256, 144), interpolation=cv2.INTER_NEAREST)
    
    hsv = cv2.cvtColor(performative_filters, cv2.COLOR_BGR2HSV)

    lower_red1 = np.array([0, 120, 70])
    upper_red1 = np.array([10, 255, 255])
    lower_red2 = np.array([170, 120, 70])
    upper_red2 = np.array([180, 255, 255])
    mask_red1 = cv2.inRange(hsv, lower_red1, upper_red1)
    mask_red2 = cv2.inRange(hsv, lower_red2, upper_red2)
    mask_red = cv2.bitwise_or(mask_red1, mask_red2)

    lower_blue = np.array([100, 150, 50])
    upper_blue = np.array([140, 255, 255])
    mask_blue = cv2.inRange(hsv, lower_blue, upper_blue)

    mask = cv2.bitwise_or(mask_red, mask_blue)

    gray = cv2.cvtColor(performative_filters, cv2.COLOR_BGR2GRAY)
    gray_bgr = cv2.cvtColor(gray, cv2.COLOR_GRAY2BGR)

    purple = np.full_like(performative_filters, (128, 0, 128)) 

    result = np.where(mask[..., None].astype(bool), purple, gray_bgr)
    
    return result

I then coded a method to get the safety state from the model:

safety_state = self._get_safety_prediction(filtered_image)

The method _get_safety_prediction is a bit more complex. It is based on the YOLO v8 model. To predict the class from a YOLO model we can use:

results = self.model.predict(filtered_image, verbose=False)

It returns a list of results, each result being a dictionary containing the class name and the confidence score.

Using YOLO methods, we can extract the class name and the confidence score from the result.

class_name = results[0].names[results[0].probs.top1]
confidence_score = results[0].probs.top1conf.item()

We finally return the safety state, which is either "safe" or "unsafe".

return "safe" if class_name.lower() == "safe" else "unsafe"

I then published the safety state to the topic:

msg = String()
msg.data = safety_state
self.publisher_.publish(msg)

💡 Tip: you can test your code by listening to the safety_state topic; something should be published every 5 seconds as a string.

Note that I was not able to see the parameters applied to the camera on the HMI. I thought that maybe human eye cannot make the difference between processed and unprocessed image as GPT, which I think is not human, decided on the parameters. But I 100% trust GPT as it has more neurons than me.

To make the conveyor run, you need to initialize it by calling a Niryo service. You just have to call it once, not each time you run the program.

ros2 service call /niryo_robot/conveyor/ping_and_set_conveyor niryo_ned_ros2_interfaces/srv/SetConveyor "{cmd: 1, id: 9}"

On the Raspberry side, make sure the Python packages opencv-python and ultralytics are installed.

Satisfactory overall job (3.5)! Even if the code is not perfect and the requests of the client are not fully fulfilled, the code is working and the robot is performing the task safely concerning the trajectory part, even if quite slowly.

Nevertheless, the quality check logic is incorrect, containing errors that can lead to unsafe vials being placed in the packaging line. Here are some feedbacks to improve the code:

Quality Check Logic

The main problem of your implementation concerns the logic of the quality check. Firstly, because the parameters you applied to the camera are not correct and lead the model to have a bad interpretation of the safety state. Indeed, the image you checked on the screen was not the one the modifications were applied to. Please review this image and apply the correct parameters to the camera, referring to the images the AI model was trained with. Optimizing the image to speed up the inference time is a good idea though.

Secondly, you decided to publish the safety state on a topic every 5 seconds. Because that's the median time to perform a trajectory in your operation, this is not a good idea. What if the image is not taken at the right time? Imagine a case where the image is taken without a vial — the last safety state would be randomly chosen.

To solve that, we recommend using a service to get the safety state. This service should be called each time the robot is in the test zone, and the safety state should be returned.

Besides, by watching the model you were given, one can clearly see that it has a lot of overfit, caused by the lack of diversity of the dataset it was trained on. This leads the model to perform correctly only if the data is taken under the same conditions as the dataset. Moreover, the model being a classification policy does not provide the possibility to detect if an object is in the image (i.e., the image will always return a safety check even if nothing is detected). In this case, it seems interesting to switch to a classification model that will return if a class is detected (here a vial) and its safety check. This will complement the service call, making sure an object is detected before returning something. You can publish the detection of an object using the quality_check/object_detected (std::bool) topic.

To solve problems coming from the model, we propose that you re-train a new YOLO model with the data of the laboratory.

💡 Tip: To train a new AI vision-based model, we can fine-tune a YOLO model. To perform that, we recommend using Roboflow. Begin by creating an account. Click on "create new project" and you should land on this page:

Here select an "Object Detection" type model, which corresponds to a segmentation model. If you still want to use a classification model, select that type instead.

The next step is to take photos to construct your dataset. You will need ~50 photos per class (safe or unsafe) to construct a solid dataset. The main goal is to avoid overfit. As you can see, we provided tools to unmount the camera from its support; this is exactly what you will need to diversify your dataset. Unlike the one provided, you have to take photos in different positions and angles of the vials. To take 50 photos in a row using the same color filter, you can write a small Python script. For instance, the script capture_img.py located in the assets folder takes 50 photos, one per second, with the correct colour filter if you run it with:

python capture_img.py <name_of_the_class>

Upload your directory on Roboflow, directly on the "Upload Data" section. Then go to annotate and start a batch. Annotate by drawing squares around the vials and naming them safe or unsafe.

Add images of both red and blue vials to the dataset and then go to the "Versions" section. Give a name to this version, do not apply preprocessing steps, but you can augment your data by applying filters; this will allow your dataset to be bigger with only 50 "real" pictures.

Once your version is ready, click on download dataset and make sure you have Yolo v8 and Show download code selected.

You should see some lines of Python code; now it is time for training! To train your own model, we recommend using this bject-detection-on-custom-dataset.ipynb" target="_blank" rel="noopener">Google Colab notebook.

Replace Step 5 with your code and execute the Colab. At the end, you will be able to perform the evaluation and validation of your model. Your model will be located at /runs/detect/train/weights/best.pt on Google Colab. Please download it, place it into the assets folder, correct the path, and adjust the code for segmentation in the detection node.

Example code:

def _get_safety_prediction(self, filtered_image):
    """Use the YOLO segmentation model to predict and analyze safe/unsafe regions"""
    if self.model is None:
        self.get_logger().warn("Model not available, returning unsafe by default")
        return "unsafe"
    
    try:
        # Run the YOLO segmentation model
        results = self.model.predict(filtered_image, verbose=False)
        
        if not results or results[0].masks is None:
            self.get_logger().warn("No segmentation mask detected")
            return "unsafe"
        
        # Extract segmentation information
        boxes = results[0].boxes
        masks = results[0].masks
        names = self.model.names
        
        safe_detected = False
        unsafe_detected = False
        
        for i, box in enumerate(boxes):
            class_id = int(box.cls[0])
            label = names[class_id].lower()
            conf = float(box.conf[0])
            
            self.get_logger().info(f"Detected {label} (conf: {conf:.2f})")
            
            if label == "unsafe":
                unsafe_detected = True
            elif label == "safe":
                safe_detected = True
        
        # Decision logic — prioritize unsafe
        if unsafe_detected:
            return "unsafe"
        elif safe_detected:
            return "safe"
        else:
            return "unsafe"
            
    except Exception as e:
        self.get_logger().error(f"Error during segmentation prediction: {e}")
        return "unsafe"

Code Quality

The client requests are not fully fulfilled in terms of performance. The professors judge that the whole architecture of the project is not optimized. The conveyor belt belonging to the robot leads to the impossibility for it to run while the robot is performing the operation.

A way to solve the problem would be to consider the robot and the conveyor belt as two different entities, each one being its own node.

Here is a scheme of the suggested architecture:

Both the conveyor belt and the robot are now independently controlled with this architecture, each being its own node. Note that the IR sensor is now owned by the conveyor belt class as the IR sensor is located on the conveyor belt, even if plugged to the robot, as its message only sends data used to control the conveyor belt.

In ROS2, there is a way to ensure that both nodes are launched in parallel, using multithreading. Each node being executed in a separate thread, we can ensure that no blocking action will stop the other node from working.

To perform this, we can use ROS2 Executors, especially the MultiThreadedExecutor, which runs each node in a separate thread.

#!/usr/bin/env python3

import rclpy
from rclpy.node import Node
from rclpy.executors import MultiThreadedExecutor
import time

class Node1(Node):
    def __init__(self):
        super().__init__("node1")
        self.timer1 = self.create_timer(1.0, self.callback_timer1)

    def callback_timer1(self):
        time.sleep(2.0)
        self.get_logger().info("cb 1")

class Node2(Node):
    def __init__(self):
        super().__init__("node2")
        self.timer2 = self.create_timer(1.0, self.callback_timer2)

    def callback_timer2(self):
        time.sleep(2.0)
        self.get_logger().info("cb 2")

def main(args=None):
    rclpy.init(args=args)

    node1 = Node1()
    node2 = Node2()

    executor = MultiThreadedExecutor()
    executor.add_node(node1)
    executor.add_node(node2)

    try:
        executor.spin()
    finally:
        executor.shutdown()
        node1.destroy_node()
        node2.destroy_node()
        rclpy.shutdown()

if __name__ == "__main__":
    main()

These two basic nodes are now able to run in parallel, each one in its own thread.

Another important concept about MultiThreadedExecutor is that by default, the callbacks are executed separately and wait for each other. This is because a default callback is a Mutually Exclusive Callback Group, meaning that only one callback can be executed at a time. This is not the expected behavior for our application, as we want the conveyor and the robot to work in parallel. Declaring a ReentrantCallbackGroup will allow the callbacks to be executed in parallel. In a Reentrant Callback Group, the callbacks are executed each time the timer calls them, meaning they are not waiting for any other callback to be executed. Please consider building your new architecture with this concept in mind.

Note that the main challenge of transforming your code to this new architecture will be to avoid deadlocks. Deadlocks occur when two or more threads are waiting for each other to release a resource, causing a halt. It is very common in multi-threaded architectures, especially when creating action clients and service clients.

In the previous code, you used rclpy.spin_until_future_complete to wait for the future to be completed. This function is not thread-safe and will cause a deadlock. Indeed, the executor is already spinning, and calling this function will cause the executor to spin again, waiting for an answer that should come in the same thread, blocking itself.

One strategy to avoid this is to use the add_done_callback method of the future, adding a callback that will be executed when the future is completed. This callback should not be executed in the same thread as the one calling add_done_callback, avoiding deadlock and freeing the thread from the waiting loop.

future = self._robot.send_goal_async(goal)
# rclpy.spin_until_future_complete(self._node, future) will cause a deadlock
future.add_done_callback(lambda fut: self._on_result_response(fut))

def _on_result_response(self, future):
    result = future.result()
    self._node.get_logger().info("Entered the callback")

Note that the callbacks should communicate their state to the main thread using a shared variable to prevent the execution of multiple actions at the same time, now that the response is received in a different thread.

Logs

The HMI is supposed to display logs to the user. The demo version you downloaded asks you to implement new publishers and subscribers to implement new logs such as cadency, time to perform a trajectory, etc.