|
import gradio as gr |
|
from client import CobotController |
|
from utils import get_credentials |
|
import json |
|
import threading |
|
import time |
|
from collections import deque |
|
import namesgenerator |
|
|
|
|
|
SESSION_TIME = 120 |
|
|
|
class QueueSystem: |
|
def __init__(self): |
|
self.queue = deque() |
|
self.current_user = None |
|
self.session_start_time = None |
|
self.lock = threading.Lock() |
|
|
|
def enqueue_user(self, user_id): |
|
with self.lock: |
|
if user_id not in self.queue and user_id != self.current_user: |
|
self.queue.append(user_id) |
|
|
|
def dequeue_user(self): |
|
with self.lock: |
|
if self.queue: |
|
return self.queue.popleft() |
|
return None |
|
|
|
def get_queue_info(self, user_id): |
|
with self.lock: |
|
if user_id == self.current_user: |
|
remaining_time = max(0, SESSION_TIME - (time.time() - self.session_start_time)) |
|
return 0, remaining_time |
|
elif user_id in self.queue: |
|
position = list(self.queue).index(user_id) + 1 |
|
if self.session_start_time: |
|
wait_time = (position - 1) * SESSION_TIME + max(0, SESSION_TIME - (time.time() - self.session_start_time)) |
|
else: |
|
wait_time = position * SESSION_TIME |
|
return position, wait_time |
|
else: |
|
return None, None |
|
|
|
def start_session(self, user_id): |
|
with self.lock: |
|
if self.current_user is None: |
|
self.current_user = user_id |
|
self.session_start_time = time.time() |
|
return True |
|
return False |
|
|
|
def end_session(self): |
|
with self.lock: |
|
if self.current_user and time.time() - self.session_start_time >= SESSION_TIME: |
|
self.current_user = None |
|
self.session_start_time = None |
|
return True |
|
return False |
|
|
|
queue_system = QueueSystem() |
|
|
|
def queue_size(): |
|
return f"There are {len(queue_system.queue)} people in the queue." |
|
|
|
|
|
def background_timer(): |
|
while True: |
|
time.sleep(1) |
|
if queue_system.end_session(): |
|
next_user = queue_system.dequeue_user() |
|
if next_user: |
|
queue_system.start_session(next_user) |
|
|
|
timer_thread = threading.Thread(target=background_timer, daemon=True) |
|
timer_thread.start() |
|
|
|
user, pwd, host, endpoint, port = get_credentials(False) |
|
client = CobotController(user, pwd, host, port, endpoint) |
|
|
|
CSS = """ |
|
#col { |
|
background-color: #161624; |
|
padding: 16px; |
|
border-radius: 8px; |
|
} |
|
#nogaprow { |
|
gap: 0px !important; |
|
} |
|
#nogapcol { |
|
padding: 0px !important; |
|
border: none !important; |
|
box-shadow: none !important; |
|
} |
|
""" |
|
|
|
""" |
|
checks the user position on the queue and returns a message along with |
|
whether the command should be executed. |
|
""" |
|
def authenticate_user(user_id): |
|
if queue_system.current_user is None: |
|
queue_system.start_session(user_id) |
|
queue_system.enqueue_user(user_id) |
|
position, wait_time = queue_system.get_queue_info(user_id) |
|
|
|
if position == 0: |
|
remaining_time_msg = f"Your turn!\nTime remaining: {wait_time:.2f} seconds." |
|
return True, remaining_time_msg |
|
elif position is not None: |
|
if position == 1: |
|
wait_msg = f"You are next!\nWait time: {wait_time:.2f} seconds." |
|
else: |
|
wait_msg = f"There are {position - 1} people ahead of you in the queue.\nWait time: {wait_time:.2f} seconds." |
|
return False, wait_msg |
|
else: |
|
return False, "Error: You are not in the queue." |
|
|
|
def enter_queue(user_id): |
|
_, msg = authenticate_user(user_id) |
|
return msg |
|
|
|
def query_angles(user_id): |
|
to_execute, queue_status_msg = authenticate_user(user_id) |
|
if to_execute: |
|
resp = client.get_angles() |
|
resp["command"] = "query/angles" |
|
return json.dumps(resp, indent=4), queue_status_msg |
|
else: |
|
return None, queue_status_msg |
|
|
|
def query_coords(user_id): |
|
to_execute, queue_status_msg = authenticate_user(user_id) |
|
if to_execute: |
|
resp = client.get_coords() |
|
resp["command"] = "query/coords" |
|
return json.dumps(resp, indent=4), queue_status_msg |
|
else: |
|
return None, queue_status_msg |
|
|
|
def query_gripper(user_id): |
|
to_execute, queue_status_msg = authenticate_user(user_id) |
|
if to_execute: |
|
resp = client.get_gripper_value() |
|
resp["command"] = "query/gripper" |
|
return json.dumps(resp, indent=4), queue_status_msg |
|
else: |
|
return None, queue_status_msg |
|
|
|
def query_camera(user_id): |
|
to_execute, queue_status_msg = authenticate_user(user_id) |
|
if to_execute: |
|
resp = client.get_camera() |
|
resp["command"] = "query/camera" |
|
if not resp["success"]: |
|
return json.dumps(resp, indent=4), None, queue_status_msg |
|
img = resp.pop("image") |
|
return json.dumps(resp, indent=4), gr.Image(visible=True, value=img), queue_status_msg |
|
else: |
|
return None, None, queue_status_msg |
|
|
|
def control_angles(user_id, angle0, angle1, angle2, angle3, angle4, angle5, movement_speed): |
|
to_execute, queue_status_msg = authenticate_user(user_id) |
|
if to_execute: |
|
resp = client.send_angles([angle0, angle1, angle2, angle3, angle4, angle5], movement_speed) |
|
resp["command"] = "control/angles" |
|
return json.dumps(resp, indent=4), queue_status_msg |
|
else: |
|
return None, queue_status_msg |
|
|
|
def control_coords(user_id, x, y, z, roll, pitch, yaw, movement_speed): |
|
to_execute, queue_status_msg = authenticate_user(user_id) |
|
if to_execute: |
|
resp = client.send_coords([x, y, z, roll, pitch, yaw], movement_speed) |
|
resp["command"] = "control/coords" |
|
return json.dumps(resp, indent=4), queue_status_msg |
|
else: |
|
return None, queue_status_msg |
|
|
|
def control_gripper(user_id, gripper_value, movement_speed): |
|
to_execute, queue_status_msg = authenticate_user(user_id) |
|
if to_execute: |
|
resp = client.send_gripper_value(gripper_value, movement_speed) |
|
resp["command"] = "control/gripper" |
|
return json.dumps(resp, indent=4), queue_status_msg |
|
else: |
|
return None, queue_status_msg |
|
|
|
def set_coords_to_current(user_id): |
|
to_execute, queue_status_msg = authenticate_user(user_id) |
|
if to_execute: |
|
resp, _ = query_coords(user_id) |
|
resp = json.loads(resp) |
|
if not resp["success"]: |
|
return None, None, None, None, None, None, queue_status_msg |
|
return resp["coords"] + [queue_status_msg] |
|
else: |
|
return None, None, None, None, None, None, queue_status_msg |
|
|
|
def set_angles_to_current(user_id): |
|
to_execute, queue_status_msg = authenticate_user(user_id) |
|
if to_execute: |
|
resp, _ = query_angles(user_id) |
|
resp = json.loads(resp) |
|
if not resp["success"]: |
|
return None, None, None, None, None, None, queue_status_msg |
|
return resp["angles"] + [queue_status_msg] |
|
else: |
|
return None, None, None, None, None, None, queue_status_msg |
|
|
|
def reset(): |
|
return 0, 0, 0, 0, 0, 0, 50 |
|
|
|
with gr.Blocks(css=CSS) as app: |
|
gr.Markdown("# MyCobot 280pi MQTT Control Demo") |
|
gr.HTML(''' |
|
<a href="https://colab.research.google.com/github/AccelerationConsortium/ac-training-lab/blob/cobot-usage-docs/src/ac_training_lab/cobot280pi/gradio-client-demo.ipynb" target="_blank"> |
|
<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" width="117" height="20"><linearGradient id="b" x2="0" y2="100%"><stop offset="0" stop-color="#bbb" stop-opacity=".1"/><stop offset="1" stop-opacity=".1"/></linearGradient><clipPath id="a"><rect width="117" height="20" rx="3" fill="#fff"/></clipPath><g clip-path="url(#a)"><path fill="#555" d="M0 0h30v20H0z"/><path fill="#007ec6" d="M30 0h87v20H30z"/><path fill="url(#b)" d="M0 0h117v20H0z"/></g><g fill="#fff" text-anchor="middle" font-family="DejaVu Sans,Verdana,Geneva,sans-serif" font-size="110"><svg x="4px" y="0px" width="22px" height="20px" viewBox="-2 0 28 24" style="background-color: #fff;border-radius: 1px;"><path style="fill:#e8710a;" d="M1.977,16.77c-2.667-2.277-2.605-7.079,0-9.357C2.919,8.057,3.522,9.075,4.49,9.691c-1.152,1.6-1.146,3.201-0.004,4.803C3.522,15.111,2.918,16.126,1.977,16.77z"/><path style="fill:#f9ab00;" d="M12.257,17.114c-1.767-1.633-2.485-3.658-2.118-6.02c0.451-2.91,2.139-4.893,4.946-5.678c2.565-0.718,4.964-0.217,6.878,1.819c-0.884,0.743-1.707,1.547-2.434,2.446C18.488,8.827,17.319,8.435,16,8.856c-2.404,0.767-3.046,3.241-1.494,5.644c-0.241,0.275-0.493,0.541-0.721,0.826C13.295,15.939,12.511,16.3,12.257,17.114z"/><path style="fill:#e8710a;" d="M19.529,9.682c0.727-0.899,1.55-1.703,2.434-2.446c2.703,2.783,2.701,7.031-0.005,9.764c-2.648,2.674-6.936,2.725-9.701,0.115c0.254-0.814,1.038-1.175,1.528-1.788c0.228-0.285,0.48-0.552,0.721-0.826c1.053,0.916,2.254,1.268,3.6,0.83C20.502,14.551,21.151,11.927,19.529,9.682z"/><path style="fill:#f9ab00;" d="M4.49,9.691C3.522,9.075,2.919,8.057,1.977,7.413c2.209-2.398,5.721-2.942,8.476-1.355c0.555,0.32,0.719,0.606,0.285,1.128c-0.157,0.188-0.258,0.422-0.391,0.631c-0.299,0.47-0.509,1.067-0.929,1.371C8.933,9.539,8.523,8.847,8.021,8.746C6.673,8.475,5.509,8.787,4.49,9.691z"/><path style="fill:#f9ab00;" d="M1.977,16.77c0.941-0.644,1.545-1.659,2.509-2.277c1.373,1.152,2.85,1.433,4.45,0.499c0.332-0.194,0.503-0.088,0.673,0.19c0.386,0.635,0.753,1.285,1.181,1.89c0.34,0.48,0.222,0.715-0.253,1.006C7.84,19.73,4.205,19.188,1.977,16.77z"/></svg><text x="245" y="140" transform="scale(.1)" textLength="30"> </text><text x="725" y="150" fill="#010101" fill-opacity=".3" transform="scale(.1)" textLength="770">Open in Colab</text><text x="725" y="140" transform="scale(.1)" textLength="770">Open in Colab</text></g> </svg> |
|
</a> |
|
''') |
|
gr.Markdown("This is a demo that uses the MQTT protocol to communicate with the MyCobot 280pi over the internet. You can remotely send commands to the cobot and query it's current status.") |
|
|
|
with gr.Row(): |
|
with gr.Column(): |
|
user_id = gr.Textbox(label="User ID", info="Enter a unique user id of your choice or take note of the automatically generated user id. This user id will be placed into a queue to give you access to the cobot, so make sure you remember it!") |
|
enter_queue_button = gr.Button("Join queue") |
|
with gr.Column(): |
|
status_text = gr.Textbox(label="Queue status", value="", interactive=False, lines=5) |
|
|
|
with gr.Row(): |
|
|
|
with gr.Column(elem_id="col"): |
|
gr.Markdown("## Query") |
|
gr.Markdown("Use buttons on this panel to query the current status of the cobot, including information like joint angles, coordinates, gripper state and what the onboard camera sees.") |
|
angle_query_button = gr.Button("Query Angles") |
|
coord_query_button = gr.Button("Query Coordinates") |
|
gripper_query_button = gr.Button("Query Gripper state") |
|
camera_query_button = gr.Button("Query Camera") |
|
|
|
|
|
with gr.Column(elem_id="col"): |
|
gr.Markdown("## Gripper Control") |
|
gr.Markdown("Use this panel to control the gripper of the cobot.") |
|
gripper_value = gr.Slider(minimum=0.0, maximum=100.0, step=1.0, label="Gripper value") |
|
speed_gripper = gr.Slider(value=50.0, minimum=0.0, maximum=100.0, step=1.0, label="Movement speed") |
|
gripper_control_button = gr.Button("Send gripper command") |
|
|
|
with gr.Row(): |
|
|
|
with gr.Column(elem_id="col"): |
|
gr.Markdown("## Angle Control") |
|
gr.Markdown("Use this panel to control the joint angles of the cobot. Each angle corresponds to one of the 6 joints on the cobot.") |
|
angle_set_button = gr.Button("Set to current angles") |
|
reset_angle_button = gr.Button("Reset angles") |
|
with gr.Row(elem_id="nogaprow"): |
|
with gr.Column(elem_id="nogapcol"): |
|
angle1 = gr.Slider(value=0.0, label="Angle 1", step=1.0, minimum=-168, maximum=168) |
|
angle3 = gr.Slider(value=0.0, label="Angle 3", step=1.0, minimum=-135, maximum=135) |
|
angle5 = gr.Slider(value=0.0, label="Angle 5", step=1.0, minimum=-150, maximum=150) |
|
with gr.Column(elem_id="nogapcol"): |
|
angle2 = gr.Slider(value=0.0, label="Angle 2", step=1.0, minimum=-145, maximum=145) |
|
angle4 = gr.Slider(value=0.0, label="Angle 4", step=1.0, minimum=-165, maximum=165) |
|
angle6 = gr.Slider(value=0.0, label="Angle 6", step=1.0, minimum=-180, maximum=180) |
|
speed_angles = gr.Slider(value=50.0, minimum=0.0, maximum=100.0, step=1.0, label="Movement speed") |
|
angle_control_button = gr.Button("Send angle command") |
|
|
|
|
|
with gr.Column(elem_id="col"): |
|
gr.Markdown("## Coordinate Control") |
|
gr.Markdown("Use this panel to control the joint coordinates of the cobot head. The angles are in [6-DoF format](https://en.wikipedia.org/wiki/Six_degrees_of_freedom).") |
|
coord_set_button = gr.Button("Set to current coords") |
|
reset_coords_button = gr.Button("Reset coordinates") |
|
with gr.Row(elem_id="nogaprow"): |
|
with gr.Column(elem_id="nogapcol"): |
|
xcoord = gr.Slider(value=0.0, label="X coordinate", step=1.0, minimum=-350, maximum=350) |
|
ycoord = gr.Slider(value=0.0, label="Y coordinate", step=1.0, minimum=-350, maximum=350) |
|
zcoord = gr.Slider(value=0.0, label="Z coordinate", step=1.0, minimum=-70, maximum=523) |
|
with gr.Column(elem_id="nogapcol"): |
|
roll = gr.Slider(value=0.0, label="Roll", step=1.0, minimum=-180, maximum=180) |
|
pitch = gr.Slider(value=0.0, label="Pitch", step=1.0, minimum=-180, maximum=180) |
|
yaw = gr.Slider(value=0.0, label="Yaw", step=1.0, minimum=-180, maximum=180) |
|
speed_coords = gr.Slider(value=50.0, minimum=0.0, maximum=100.0, step=1.0, label="Movement speed") |
|
coord_control_button = gr.Button("Send coordinate command") |
|
|
|
response = gr.Textbox(label="Response") |
|
response_image = gr.Image(visible=False) |
|
|
|
|
|
angle_query_button.click( |
|
query_angles, |
|
inputs = [user_id], |
|
outputs = [response, status_text] |
|
) |
|
coord_query_button.click( |
|
query_coords, |
|
inputs = [user_id], |
|
outputs = [response, status_text] |
|
) |
|
gripper_query_button.click( |
|
query_gripper, |
|
inputs = [user_id], |
|
outputs = [response, status_text] |
|
) |
|
camera_query_button.click( |
|
query_camera, |
|
inputs = [user_id], |
|
outputs = [response, response_image, status_text] |
|
) |
|
gripper_control_button.click( |
|
control_gripper, |
|
inputs = [user_id, gripper_value, speed_gripper], |
|
outputs = [response, status_text] |
|
) |
|
angle_control_button.click( |
|
control_angles, |
|
inputs = [user_id, angle1, angle2, angle3, angle4, angle5, angle6, speed_angles], |
|
outputs = [response, status_text] |
|
) |
|
coord_control_button.click( |
|
control_coords, |
|
inputs = [user_id, xcoord, ycoord, zcoord, roll, pitch, yaw, speed_coords], |
|
outputs = [response, status_text] |
|
) |
|
coord_set_button.click( |
|
set_coords_to_current, |
|
inputs = [user_id], |
|
outputs = [xcoord, ycoord, zcoord, roll, pitch, yaw, status_text] |
|
) |
|
angle_set_button.click( |
|
set_angles_to_current, |
|
inputs = [user_id], |
|
outputs = [angle1, angle2, angle3, angle4, angle5, angle6, status_text] |
|
) |
|
reset_angle_button.click( |
|
reset, |
|
outputs = [angle1, angle2, angle3, angle4, angle5, angle6, speed_angles] |
|
) |
|
reset_coords_button.click( |
|
reset, |
|
outputs = [xcoord, ycoord, zcoord, roll, pitch, yaw, speed_coords] |
|
) |
|
enter_queue_button.click( |
|
enter_queue, |
|
inputs = [user_id], |
|
outputs = [status_text] |
|
) |
|
|
|
app.load( |
|
namesgenerator.get_random_name, |
|
outputs=[user_id] |
|
) |
|
|
|
app.load( |
|
queue_size, |
|
outputs=[status_text] |
|
) |
|
|
|
app.queue(default_concurrency_limit=1, max_size=100) |
|
app.launch() |
|
|
