(3.6-orientation)= # 🧩 3.6 Solid Sample Transfer ```{contents} :depth: 3 ``` ## 🔰 Tutorial This module revolves around "transfer of solid samples", and completes complex automation tasks through ROS, AprilTags, multi-axis robots and workflow orchestration platform. The goal is to learn how to coordinate different modules (including robot control, visual recognition, and workflow management) to complete complex solid sample transfer processes, such as performing a grabbing task from one workstation to another for sample processing (such as screwing a bottle cap or dropping a liquid). This is a multi-step automation process involving the coordinated operation of multiple hardware devices and software platforms. ### System Overview - MyCobot 280 Pi: For precise sample handling and manipulation - MyAGV (Automated Guided Vehicle): For transporting samples between stations - Liquid Handler: For adding liquids to samples - Powder Dispenser: For adding powders to samples - AprilTags: For identifying and locating different stations and samples - ROS2 Humble: For robot control and system integration - Prefect: For workflow orchestration ### Bill of Materials - [MyCobot Pi - World's Smallest and Lightest Six-Axis Collaborative Robot](https://shop.elephantrobotics.com/en-ca/collections/mycobot-280/products/mycobot-pi-worlds-smallest-and-lightest-six-axis-collaborative-robot) A compact and versatile six-axis robot suitable for performing precise sample transfer tasks. - [Camera Flange 2.0](https://shop.elephantrobotics.com/en-ca/collections/camera-modules/products/camera-flange-2-0) A camera module attachment for enhanced visual feedback during sample transfer operations. - [Adaptive Gripper](https://shop.elephantrobotics.com/en-ca/collections/grippers/products/adaptive-gripper) A flexible gripper designed for secure handling and transfer of solid samples during automated processes. - [G-Shape Base 2.0](https://shop.elephantrobotics.com/en-ca/collections/fixed-bases/products/g-shape-base-2-0) A stable base for mounting the MyCobot robot, ensuring precision during robotic movements and sample handling. - [Multi-Axis Robot](https://shop.elephantrobotics.com/en-ca/collections/mycobot-280/products/mycobot-pi-worlds-smallest-and-lightest-six-axis-collaborative-robot) A robotic arm designed for executing precise sample movements across multiple axes. - Printed AprilTags (can be generated and printed from [AprilTag Generation](https://github.com/AprilRobotics/apriltag-generation)) -- [liquid handler] -- [powder dispenser] ### Required Software - [Prefect](https://www.prefect.io/) A workflow orchestration tool to manage and coordinate asynchronous tasks during the sample transfer process. - [AprilTags Python Library](https://pypi.org/project/apriltag/) Used for spatial referencing and tracking of sample containers using AprilTags for accurate positioning. - [ROS2 Humble](https://docs.ros.org/en/humble/Installation.html) ROS2 (Robot Operating System) is an open-source framework for building robot applications. **Humble Hawksbill** is the currently recommended version due to its stability and long-term support. - [Ubuntu 22.04 LTS](https://releases.ubuntu.com/22.04/) Ubuntu 22.04 LTS is the recommended Linux distribution for running ROS2, as it provides a stable and compatible environment. **Ubuntu 24.04** (the latest version) may encounter compatibility issues with ROS2 and other tools. #### Documentation - [MyCobot Pi Documentation](https://docs.elephantrobotics.com/docs/gitbook-en/2-serialproduct/2.1-280/2.1.2-PI.html) A guide for setting up and controlling the MyCobot robot for sample transfer tasks. - [Gripper Control via Python](https://docs.elephantrobotics.com/docs/gitbook-en/7-ApplicationBasePython/7.5_gripper.html) Detailed instructions for using Python to control the adaptive gripper during the sample transfer process. ### Notes These materials enable the setup and execution of a solid sample transfer process, utilizing ROS, AprilTags, and a multi-axis robot for precision control. Prefect orchestrates the workflow, allowing for asynchronous task management in a highly automated and efficient manner. This setup is ideal for educational and research environments that require robust and flexible sample handling solutions. If you need to review ROS and robotic arm operations, please refer back to Section 3.4-Mobile-robotics for a detailed refresher on these topics. --- ### Analysis of each module and its combination: 1. ROS (Robot Operating System): - Introduction to ROS: ROS is an open source framework for robot development and control, suitable for managing complex robot operations, especially multi-axis robots. In the solid sample transfer scenario, ROS is used to control multi-axis robots to complete precise sample grabbing, movement, and manipulation. - Application scenario: ROS is responsible for controlling the multi-axis robot to grab samples from workstation A and transfer them to workstation B. At the same time, ROS can obtain real-time feedback by connecting different sensors (such as cameras and force sensors) to ensure the accuracy of the operation. - How to combine with other modules: - Integration with AprilTags: ROS is integrated with AprilTags to identify the position and orientation of the sample or target workstation, providing the robot with accurate coordinates for grabbing and placing. - Integrate with workflow orchestration platform: ROS is responsible for low-level robot control, while task scheduling and sequence management are handled by workflow orchestration platforms (such as Prefect). 2. AprilTags (Tag Identification): - About AprilTags: AprilTags is a two-dimensional label system, similar to QR code, commonly used for position and posture estimation in robot vision. In this scenario, AprilTags can be attached to samples, containers or workstations to help robots accurately locate and operate. - Application scenario: The multi-axis robot needs to identify the AprilTags on the sample through a visual system (such as a camera) to determine the specific position and posture of the sample to ensure accurate grasping and operation. - How to combine with other modules: - Integration with ROS: By publishing the detection results of AprilTags through ROS, the robot can adjust its grasping posture and position in real time. For example, after detecting the AprilTags of a sample or container, ROS can adjust the operation of the robot arm based on the visual input. - Combined with multi-axis robots: AprilTags provide precise sample location information, ensuring that multi-axis robots can accurately grasp and place samples in complex environments. 3. Multi-axis robot: - Introduction to multi-axis robots: Multi-axis robots have multiple degrees of freedom and can perform complex operations in three-dimensional space, such as grasping, rotating, unscrewing bottle caps, pouring samples, etc. In this scenario, the robot needs to complete a complete set of operations from grasping samples, moving samples, to accurately placing samples. - Application scenarios: Multi-axis robots use ROS instructions to complete complex sample transfer tasks, such as grabbing solid samples, unscrewing or tightening bottle caps, and accurately placing samples at the target location. - How to combine with other modules: - Integration with ROS: Through commands issued by ROS, the multi-axis robot can perform various movements, including linear movement, rotation, grasping, etc. - Combined with AprilTags: The robot detects AprilTags to determine the location of the sample or workstation and perform accurate operations. 4. Workflow orchestration platform (such as Prefect): - Introduction to Workflow Orchestration: Workflow orchestration platforms (such as Prefect) can be used to manage and schedule complex asynchronous tasks, ensure that the various steps of robot operations are executed in sequence, and handle concurrent tasks that may arise. - Application scenario: During the entire process of solid sample transfer, the workflow orchestration platform is responsible for managing the execution order of each step. For example, the steps of a task include: 1) detecting the sample position, 2) grabbing the sample, 3) transferring the sample to the next workstation, 4) completing the specified operation (such as unscrewing the bottle cap), and 5) putting the sample back in place. - How to combine with other modules: - Integration with ROS: ROS is responsible for executing specific robot control commands, while the workflow orchestration platform is responsible for determining the order and dependencies of tasks to ensure that each task is triggered at the right time. - Integrated with AprilTags: In the task process, the workflow orchestration platform can trigger subsequent tasks based on the detection results of AprilTags (such as identifying the sample location). ### Task combination and process #### Application scenario: Solid sample transfer process ### Workflow Description 1. Sample pickup from initial station 2. Transfer to liquid handling station 3. Liquid addition 4. Transfer to powder dispensing station 5. Powder addition 6. Transfer to final station #### 1. Station Identification with AprilTags AprilTags are used to identify different stations in our setup. Here's an example of how to detect AprilTags: ```python import cv2 from pupil_apriltags import Detector def detect_apriltags(image): gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) detector = Detector(families="tag36h11") results = detector.detect(gray) for r in results: # Extract tag ID and pose information tag_id = r.tag_id pose = r.pose_t print(f"Detected AprilTag ID: {tag_id} at position: {pose}") # Usage cap = cv2.VideoCapture(0) ret, frame = cap.read() if ret: detect_apriltags(frame) cap.release() ``` #### 2. MyCobot Control for Sample Manipulation Here's an example of how to control the MyCobot to pick up a sample: ```python from pymycobot.mycobot import MyCobot def pickup_sample(mycobot, x, y, z): # Move to the sample position mycobot.send_coords([x, y, z, 0, 0, 0], 50, 0) # Close the gripper mycobot.set_gripper_state(0, 50) # Lift the sample mycobot.send_coords([x, y, z + 50, 0, 0, 0], 50, 0) # Usage mycobot = MyCobot("/dev/ttyAMA0") pickup_sample(mycobot, 100, 100, 50) ``` #### 3. AGV Navigation between Stations Here's a simple example of AGV navigation using ROS: ```python import rospy from geometry_msgs.msg import Twist def move_agv(linear_speed, angular_speed, duration): pub = rospy.Publisher("/cmd_vel", Twist, queue_size=10) rospy.init_node("agv_controller", anonymous=True) rate = rospy.Rate(10) # 10hz twist = Twist() twist.linear.x = linear_speed twist.angular.z = angular_speed start_time = rospy.Time.now() while (rospy.Time.now() - start_time).to_sec() < duration: pub.publish(twist) rate.sleep() # Usage move_agv(0.2, 0, 5) # Move forward for 5 seconds ``` #### 4. Liquid Handler Integration Here's a hypothetical example of controlling a liquid handler: ```python class LiquidHandler: def __init__(self): # Initialize liquid handler pass def dispense_liquid(self, volume_ml): print(f"Dispensing {volume_ml}ml of liquid") # Code to control the liquid handler pass # Usage liquid_handler = LiquidHandler() liquid_handler.dispense_liquid(5) ``` #### 5. Powder Dispenser Integration Similarly, here's a hypothetical example for a powder dispenser: ```python class PowderDispenser: def __init__(self): # Initialize powder dispenser pass def dispense_powder(self, weight_g): print(f"Dispensing {weight_g}g of powder") # Code to control the powder dispenser pass # Usage powder_dispenser = PowderDispenser() powder_dispenser.dispense_powder(2) ``` #### 6. Workflow Orchestration with Prefect Now, we can use Prefect to orchestrate all these steps: ```python from prefect import task, Flow @task def identify_station(): # Use the AprilTag detection code here pass @task def pickup_sample(): # Use the MyCobot control code here pass @task def move_to_liquid_station(): # Use the AGV navigation code here pass @task def add_liquid(): liquid_handler = LiquidHandler() liquid_handler.dispense_liquid(5) @task def move_to_powder_station(): # Use the AGV navigation code here pass @task def add_powder(): powder_dispenser = PowderDispenser() powder_dispenser.dispense_powder(2) @task def move_to_final_station(): # Use the AGV navigation code here pass @task def place_sample(): # Use the MyCobot control code here to place the sample pass with Flow("Solid Sample Transfer and Processing") as flow: station = identify_station() sample = pickup_sample() to_liquid = move_to_liquid_station(upstream=sample) liquid = add_liquid(upstream=to_liquid) to_powder = move_to_powder_station(upstream=liquid) powder = add_powder(upstream=to_powder) to_final = move_to_final_station(upstream=powder) placed = place_sample(upstream=to_final) # Run the flow flow.run() ``` This structure provides individual code examples for each step, making it easier to understand and implement each component separately. The final Prefect workflow then brings all these steps together into a cohesive process. ## Extending the System and Troubleshooting This tutorial provides a basic framework for automated solid sample transfer. However, real-world applications often require more complex setups and face various challenges. Here are some considerations for extending the system and troubleshooting common issues: ### 1. Scalability: - For larger-scale operations, consider implementing a multi-robot system with centralized control. - Utilize a database to track sample information and processing history. This example code snippet demonstrates basic CRUD (Create, Read, Update, Delete) operations using pymongo, the Python driver for MongoDB. Here's what each section does: - Connection: We establish a connection to the MongoDB server. Replace 'mongodb://localhost:27017/' with your actual MongoDB connection string. Database and Collection Selection: We select the database and collection we want to work with. Replace 'your_database_name' and 'your_collection_name' with your actual database and collection names. - Insert Operation: We insert a new document into the collection and print the ID of the inserted document. - Query Operation: We search for a document in the collection based on a specific query and print the result. Update Operation: We update a document in the collection, changing the 'data' field, and print the number of modified documents. 1. Delete Operation: We delete a document from the collection based on a query and print the number of deleted documents. 2. Connection Closure: Finally, we close the connection to the MongoDB server. This snippet provides a basic overview of working with MongoDB using pymongo. In a real-world scenario, you would typically include error handling and possibly use context managers for the database connection. Also, ensure that you have the necessary permissions to perform these operations on your MongoDB instance. ```python import pymongo from pymongo import MongoClient # Connect to MongoDB client = MongoClient( "mongodb://localhost:27017/" ) # Replace with your MongoDB connection string # Select database and collection db = client["your_database_name"] collection = db["your_collection_name"] # Insert a document document = {"course_id": "DEMO101", "data": "Some sample data"} insert_result = collection.insert_one(document) print(f"Inserted document ID: {insert_result.inserted_id}") # Query a document query = {"course_id": "DEMO101"} result = collection.find_one(query) print(f"Found document: {result}") # Update a document update_query = {"course_id": "DEMO101"} new_values = {"$set": {"data": "Updated sample data"}} update_result = collection.update_one(update_query, new_values) print(f"Modified {update_result.modified_count} document(s)") # Delete a document delete_query = {"course_id": "DEMO101"} delete_result = collection.delete_one(delete_query) print(f"Deleted {delete_result.deleted_count} document(s)") # Close the connection client.close() ``` ### Tips When dealing with multiple robots, deciding which parts should be implemented in ROS2 and which in Prefect is indeed an important consideration. Here are some factors to consider and recommendations: ROS2 (Robot Operating System 2): 1. Real-time control and communication: ROS2 is better suited for handling real-time control and low-latency communication needs of robots. Robot-specific functionalities: Tasks such as navigation, motion planning, and sensor data processing should be implemented in ROS2. Inter-robot collaboration: If multiple robots need to communicate directly or coordinate actions with each other, this is typically best handled within ROS2. 4. Hardware abstraction: ROS2 provides a good hardware abstraction layer, allowing the same code to run on different robot platforms. Prefect: Workflow orchestration: Prefect is more suitable for handling high-level task orchestration and workflow management. Long-term task scheduling: For tasks that need to run for extended periods or execute periodically, Prefect is a better choice. Data processing pipelines: If complex data processing or analysis workflows are involved, these can be implemented in Prefect. 4. Integration with external systems: Prefect may be more suitable for handling integration with databases, cloud services, or other external systems. 5. Error handling and retry mechanisms: Prefect provides robust error handling and task retry capabilities, suitable for handling long-running tasks that may fail. Recommendations for combined use: Use ROS2 to handle low-level control and real-time operations for each robot. 2. Use Prefect to manage high-level tasks and workflows across multiple robots. 3. Create a ROS2 node to act as a bridge between Prefect and the ROS2 system, handling command dispatching and status reporting. Implement long-running data collection, analysis, and reporting tasks in Prefect. Use Prefect to manage complex task sequences that require collaboration between multiple robots. Example scenario: Consider an automated warehouse system with multiple robots: In ROS2: Implement navigation, item grasping, obstacle avoidance, etc., for each robot. In Prefect: Manage the overall workflow, such as task allocation, progress tracking, data collection, and report generation. By using this approach, you can take full advantage of ROS2's real-time performance and robot control capabilities, while leveraging Prefect's workflow management and task orchestration features to handle higher-level system coordination. ### 2. Error Handling: - Implement robust error handling and recovery mechanisms. - Use sensors to detect collisions or unexpected obstacles. - Develop a logging system to record all operations and errors for later analysis. Let's take the liquid handling station as an example to demonstrate error handling and logging: ```python import logging from prefect import task from datetime import timedelta class LiquidHandlerError(Exception): pass class LiquidHandler: def __init__(self): self.logger = logging.getLogger(__name__) # Logger setup code... def dispense_liquid(self, volume_ml: float) -> None: try: self._check_volume(volume_ml) self._check_liquid_level() self._dispense(volume_ml) self.logger.info(f"Successfully dispensed {volume_ml}ml of liquid") except LiquidHandlerError as e: self.logger.error(f"Error dispensing liquid: {str(e)}") raise # Other methods like _check_volume, _check_liquid_level, _dispense... @task(max_retries=3, retry_delay=timedelta(seconds=5)) def add_liquid(volume_ml: float): liquid_handler = LiquidHandler() try: liquid_handler.dispense_liquid(volume_ml) return "Liquid added successfully" except LiquidHandlerError as e: liquid_handler.logger.error(f"Failed to add liquid after retries: {str(e)}") return None ``` In this example, we've implemented several key error handling strategies: - Custom Exception: We define `LiquidHandlerError` for specific errors related to liquid handling. - Logging: We use Python's `logging` module to record operations and errors. This helps in debugging and monitoring the system. - Error Checks: In the `dispense_liquid` method, we perform checks for valid volume and sufficient liquid level before dispensing. - Try-Except Blocks: We use try-except to catch and handle errors, ensuring that errors are logged and don't crash the entire system. - Prefect Task with Retries: We use Prefect's `@task` decorator with `max_retries` and `retry_delay` parameters. This allows the system to automatically retry failed operations, which can help overcome temporary issues. By implementing these error handling mechanisms, we can create a more robust and reliable automated system. This approach helps in: - Identifying and logging specific errors for easier troubleshooting. - Preventing system crashes due to unexpected errors. - Automatically recovering from temporary failures through retries. - Providing clear feedback about the success or failure of operations. Similar error handling strategies can be applied to other stations in the automated lab, such as the powder dispenser or the robotic arm. Each component should have its own error checks, logging, and recovery mechanisms tailored to its specific operations and potential failure modes. Another example of a human-in-the-loop workflow orchestration using Prefect, demonstrating how to handle an error by triggering an operator to check it out: ```python from prefect import task, flow from prefect.tasks import task_input_hash from datetime import timedelta import time import random @task(cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=1)) def robot_task(task_id): # Simulate a robot task that might fail if random.random() < 0.3: # 30% chance of failure raise Exception(f"Robot task {task_id} failed") time.sleep(2) # Simulate task duration return f"Task {task_id} completed successfully" @task def notify_operator(task_id): print(f"ALERT: Task {task_id} has failed. An operator needs to check the robot.") # In a real scenario, this could send an email, SMS, or trigger an alert system operator_response = input("Has the issue been resolved? (yes/no): ") return operator_response.lower() == "yes" @flow def robot_workflow(): tasks = [robot_task.submit(i) for i in range(5)] for i, task in enumerate(tasks): try: result = task.result() print(result) except Exception as e: print(f"Error in task {i}: {str(e)}") resolved = False while not resolved: resolved = notify_operator(i) if not resolved: print("Please resolve the issue before continuing.") time.sleep(5) # Wait before checking again # Retry the task after the operator has resolved the issue retry_result = robot_task(i) print(f"Retry result: {retry_result}") if __name__ == "__main__": robot_workflow() ``` This example demonstrates: - A robot_task that simulates a robot operation with a 30% chance of failure. - A notify_operator task that alerts an operator when a task fails and waits for confirmation that the issue has been resolved. - A robot_workflow that: - Submits multiple robot tasks - Checks the results of each task If a task fails, it notifies an operator and waits for confirmation before retrying the task In a real-world scenario, you would replace the print statements and input function with actual notification systems (e.g., email, SMS, or integration with an alerting platform) and a more sophisticated method for the operator to confirm that the issue has been resolved (e.g., through a web interface or mobile app). This workflow demonstrates how Prefect can be used to: - Orchestrate multiple robot tasks - Handle errors gracefully - Incorporate human intervention when needed - Retry tasks after issues have been resolved This approach ensures that the workflow can continue even when unexpected issues occur, by bringing in human expertise to resolve problems that the automated system can't handle on its own. ### 3. Precision and Accuracy: - Regularly calibrate the robot and sensors to maintain high accuracy. - Consider using computer vision techniques for more precise sample identification and positioning. ### 4. Safety: - Implement emergency stop procedures accessible through both hardware and software. - Use light curtains or other safety sensors to prevent human-robot collisions. ### 5. Flexibility: - Design modular software components to easily add or modify processing steps. - Create a configuration system to adjust parameters without changing the core code. ### 6. Integration with Laboratory Information Management Systems (LIMS): - Develop APIs to connect your automated system with existing LIMS for seamless data flow. Here's an example of how you might integrate your automated system with a cloud-based LIMS using AWS DynamoDB: ```python import boto3 from prefect import task class CloudLIMS: def __init__(self): self.dynamodb = boto3.resource("dynamodb") self.table = self.dynamodb.Table("SampleData") def log_sample_data(self, sample_id, operation, result): self.table.put_item( Item={ "SampleID": sample_id, "Operation": operation, "Result": result, "Timestamp": datetime.now().isoformat(), } ) @task def process_sample(sample_id): lims = CloudLIMS() # Perform sample processing steps... # Log the results to the cloud LIMS lims.log_sample_data(sample_id, "LiquidAddition", "Success") lims.log_sample_data(sample_id, "PowderAddition", "Success") return f"Sample {sample_id} processed successfully" # Include this task in your Prefect flow with Flow("Sample Processing") as flow: sample_id = "SAMPLE001" result = process_sample(sample_id) # Run the flow flow_state = flow.run() ``` This example demonstrates how to: - Use AWS DynamoDB as a cloud-based LIMS database - Create a `CloudLIMS` class to handle interactions with the database - Integrate LIMS operations into your Prefect workflow By using a cloud-based solution, you can: - Store and access sample data from anywhere - Scale your data storage as your operation grows - Implement data redundancy and backup strategies - Enable real-time data sharing and collaboration Remember to handle authentication and security appropriately when working with cloud services, and ensure compliance with any relevant data protection regulations. ### 7. Future Enhancements: - Explore machine learning techniques for adaptive robot control and anomaly detection. - Investigate the use of collaborative robots (cobots) for safer human-robot interaction. - Consider implementing augmented reality interfaces for easier system monitoring and control. By considering these aspects, you can create a more robust, flexible, and scalable automated sample transfer system suitable for a wide range of laboratory and industrial applications. ## 🚀 Quiz ::::{tab-set} :sync-group: category :::{tab-item} W 2024 :sync: w2024 ::: :::{tab-item} Sp/Su 2024 :sync: sp2024 https://q.utoronto.ca/courses/370069/assignments/1393650 ::: :::{tab-item} Sp/Su 2025 :sync: sp2025 https://q.utoronto.ca/courses/393350/assignments/1499713?display=full_width_with_nav ::: :::: ## 📄 Assignment ::::{tab-set} :sync-group: category :::{tab-item} W 2024 :sync: w2024 ::: :::{tab-item} Sp/Su 2024 :sync: sp2024 https://q.utoronto.ca/courses/370069/assignments/1394534 ::: :::{tab-item} Sp/Su 2025 :sync: sp2025 https://q.utoronto.ca/courses/393350/assignments/1499714?display=full_width_with_nav ::: :::: ## Course Completion For those who have completed the course, thank you for your participation, and congratulations! 🥳 Once you have completed all the modules, fill out the [Certificate of Completion Request form](https://learn.utoronto.ca/help/forms-and-applications/confirmation-completion-request) through the School of Continuing Studies to receive your microcredential. This process may take up to 4-6 weeks. As a reminder, 70% is the required threshold to pass. Please also consider filling out [the final course evaluation form](https://forms.office.com/r/DnUwydGgsb) listed on your course syllabus as well as [the learner equity survey](https://redcap.utoronto.ca/surveys/?s=77R3YNHDHX3TKRLD). Once you have received your microcredential, consider adding the corresponding badge your LinkedIn profile to promote your accomplishment. Likewise, we would love to see a social media post from you using the `#AcMicrocourses` hashtag with the Acceleration Consortium account tagged ([`@acceleration-consortium`](https://www.linkedin.com/company/acceleration-consortium/) for LinkedIn and [`@acceleration_c`](https://x.com/acceleration_c) for X).