checkpoint

app.py

@@ -1,244 +1,329 @@
+import gradio as gr
+from client import CobotController
+from utils import get_credentials
+from typing import Any
+import json
+
+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;  /* Removes the border around each column */
+    box-shadow: none !important;
+}
+"""
+
+def handle_request(
+	command, gripper_value, gripper_speed, angle1, angle2, angle3, angle4,
+	angle5, angle6, angle_speed, xcoord, ycoord, zcoord, roll, pitch, yaw, coord_speed,
+):
+	if command == "query/angles":
+		resp = client.get_angles()
+		resp["command"] = command
+		return json.dumps(resp, indent=4), gr.Image(visible=False)
+	elif command == "query/coords":
+		resp = client.get_coords()
+		resp["command"] = command
+		return json.dumps(resp, indent=4), gr.Image(visible=False)
+	elif command == "query/camera":
+		resp = client.get_camera()
+		resp["command"] = command
+		if not resp["success"]:
+			return json.dumps(resp, indent=4)
+		img = resp.pop("image")
+		return json.dumps(resp, indent=4), gr.Image(visible=True, value=img)
+	elif command == "query/gripper":
+		resp = client.get_gripper_value()
+		resp["command"] = command
+		return json.dumps(resp, indent=4), gr.Image(visible=False)
+	elif command == "control/angles":
+		resp = client.send_angles(
+			angle_list = [angle1, angle2, angle3, angle4, angle5, angle6],
+			speed = angle_speed
+		)
+		resp["command"] = command
+		return json.dumps(resp, indent=4), gr.Image(visible=False)
+	elif command == "control/gripper":
+		resp = client.send_gripper_value(
+			value = gripper_value,
+			speed = gripper_speed
+		)
+		resp["command"] = command
+		return json.dumps(resp, indent=4), gr.Image(visible=False)
+	elif command == "control/coords":
+		resp = client.send_coords(
+			coord_list = [xcoord, ycoord, zcoord, roll, pitch, yaw],
+			speed = coord_speed
+		)
+		resp["command"] = command
+		return json.dumps(resp, indent=4), gr.Image(visible=False)
+
+with gr.Blocks(css=CSS) as app:
+	gr.Markdown("# MyCobot 280pi MQTT Control Demo")
+	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.")
+
+	user_id = gr.Textbox(label="User ID", info="Enter a unique user id of your choice. This user id will be placed into a queue to give you access to the cobot, so make sure you remember it!")
+
+	with gr.Row():
+		# QUERY PANEL
+		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")
+
+		# GRIPPER PANEL
+		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", info="0 = gripper fully closed, 100 = gripper fully open")
+			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():
+		# ANGLE PANEL
+		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.")
+			with gr.Row(elem_id="nogaprow"):
+				with gr.Column(elem_id="nogapcol"):
+					angle1 = gr.Number(value=0.0, label="Angle 1", step=5.0)
+					angle3 = gr.Number(value=0.0, label="Angle 3", step=5.0)
+					angle5 = gr.Number(value=0.0, label="Angle 5", step=5.0)
+				with gr.Column(elem_id="nogapcol"):
+					angle2 = gr.Number(value=0.0, label="Angle 2", step=5.0)
+					angle4 = gr.Number(value=0.0, label="Angle 4", step=5.0)
+					angle6 = gr.Number(value=0.0, label="Angle 6", step=5.0)
+			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")
+
+		# COORD PANEL
+		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).")
+			with gr.Row(elem_id="nogaprow"):
+				with gr.Column(elem_id="nogapcol"):
+					xcoord = gr.Number(value=0.0, label="X coordinate", step=5.0)
+					ycoord = gr.Number(value=0.0, label="Y coordinate", step=5.0)
+					zcoord = gr.Number(value=0.0, label="Z coordinate", step=5.0)
+				with gr.Column(elem_id="nogapcol"):
+					roll = gr.Number(value=0.0, label="Roll", step=5.0)
+					pitch = gr.Number(value=0.0, label="Pitch", step=5.0)
+					yaw = gr.Number(value=0.0, label="Yaw", step=5.0)
+			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(
+		handle_request,
+		inputs = [
+			gr.Textbox(value="query/angles", visible=False),
+			gripper_value, speed_gripper, angle1, angle2, angle3, angle4,
+			angle5, angle6, speed_angles, xcoord, ycoord, zcoord, roll, pitch,
+			yaw, speed_coords 
+		],
+		outputs = [response, response_image]
+	)
+	coord_query_button.click(
+		handle_request,
+		inputs = [
+			gr.Textbox(value="query/coords", visible=False),
+			gripper_value, speed_gripper, angle1, angle2, angle3, angle4,
+			angle5, angle6, speed_angles, xcoord, ycoord, zcoord, roll, pitch,
+			yaw, speed_coords 
+		],
+		outputs = [response, response_image]
+	)
+	gripper_query_button.click(
+		handle_request,
+		inputs = [
+			gr.Textbox(value="query/gripper", visible=False),
+			gripper_value, speed_gripper, angle1, angle2, angle3, angle4,
+			angle5, angle6, speed_angles, xcoord, ycoord, zcoord, roll, pitch,
+			yaw, speed_coords 
+		],
+		outputs = [response, response_image]
+	)
+	camera_query_button.click(
+		handle_request,
+		inputs = [
+			gr.Textbox(value="query/camera", visible=False),
+			gripper_value, speed_gripper, angle1, angle2, angle3, angle4,
+			angle5, angle6, speed_angles, xcoord, ycoord, zcoord, roll, pitch,
+			yaw, speed_coords 
+		],
+		outputs = [response, response_image]
+	)
+	gripper_control_button.click(
+		handle_request,
+		inputs = [
+			gr.Textbox(value="control/gripper", visible=False),
+			gripper_value, speed_gripper, angle1, angle2, angle3, angle4,
+			angle5, angle6, speed_angles, xcoord, ycoord, zcoord, roll, pitch,
+			yaw, speed_coords 
+		],
+		outputs = [response, response_image]
+	)
+	angle_control_button.click(
+		handle_request,
+		inputs = [
+			gr.Textbox(value="control/angles", visible=False),
+			gripper_value, speed_gripper, angle1, angle2, angle3, angle4,
+			angle5, angle6, speed_angles, xcoord, ycoord, zcoord, roll, pitch,
+			yaw, speed_coords 
+		],
+		outputs = [response, response_image]
+	)
+	coord_control_button.click(
+		handle_request,
+		inputs = [
+			gr.Textbox(value="control/coords", visible=False),
+			gripper_value, speed_gripper, angle1, angle2, angle3, angle4,
+			angle5, angle6, speed_angles, xcoord, ycoord, zcoord, roll, pitch,
+			yaw, speed_coords 
+		],
+		outputs = [response, response_image]
+	)
+
+app.launch()
+
 # import gradio as gr
-# from client import CobotController
-# from utils import get_credentials
-# import json
-
-# options = [
-# 	"query/angles", "query/coords", "query/gripper", "query/camera",
-# 	"control/angles", "control/coords", "control/gripper"
-# ]
-
-# with gr.Blocks() as app:
-# 	user, pwd, host, endpoint, port = get_credentials(False)
-# 	client = CobotController(user, pwd, host, port, endpoint)
-
-# 	def handle_request(
-# 		selected_command, angle1=0.0, angle2=0.0, angle3=0.0, angle4=0.0, 
-# 		angle5=0.0, angle6=0.0, coord1=0.0, coord2=0.0, coord3=0.0, coord4=0.0, 
-# 		coord5=0.0, coord6=0.0, gripper_value=0.0, speed_angles=0.0, speed_coords=0.0,
-# 		speed_gripper=0.0
-# 	):
-# 		img = None
-# 		if selected_command == "query/angles":
-# 			response = client.get_angles()
-# 		elif selected_command == "query/coords":
-# 			response = client.get_coords()
-# 		elif selected_command == "query/gripper":
-# 			response = client.get_gripper_value()
-# 		elif selected_command == "query/camera":
-# 			response = client.get_camera()
-# 			img = response.pop("image")
-# 		elif selected_command == "control/angles":
-# 			response = client.send_angles(
-# 				angle_list = [angle1, angle2, angle3, angle4, angle5, angle6],
-# 				speed = speed_angles
-# 			)
-# 		elif selected_command == "control/coords":
-# 			response = client.send_coords(
-# 				angle_list = [coord1, coord2, coord3, coord4, coord5, coord6],
-# 				speed = speed_coords
-# 			)
-# 		elif selected_command == "control/gripper":
-# 			response = client.send_gripper_value(
-# 				value = gripper_value,
-# 				speed = speed_gripper
-# 			)
-
-# 		text_output = json.dumps(response, indent=4)
-# 		if img is not None:
-# 			return text_output, gr.update(value=img, visible=True)
-# 		return text_output, None
-
-# 	gr.Markdown("# MyCobot 280pi MQTT Control demo")
-# 	gr.Markdown("This is a public demo of the MyCobot 280pi...")
-
-# 	selected_command = gr.Dropdown(choices=options, label="Select command")
-
-# 	with gr.Group(visible=False) as query_angles_group:
-# 		pass
-
-# 	with gr.Group(visible=False) as query_coords_group:
-# 		pass
-
-# 	with gr.Group(visible=False) as query_camera_group:
-# 		pass
-
-# 	with gr.Group(visible=False) as query_gripper_group:
-# 		pass
-
-# 	with gr.Group(visible=False) as control_angles_group:
-# 		angle1 = gr.Number(value=0.0, label="Angle 1", step=5.0)
-# 		angle2 = gr.Number(value=0.0, label="Angle 2", step=5.0)
-# 		angle3 = gr.Number(value=0.0, label="Angle 3", step=5.0)
-# 		angle4 = gr.Number(value=0.0, label="Angle 4", step=5.0)
-# 		angle5 = gr.Number(value=0.0, label="Angle 5", step=5.0)
-# 		angle6 = gr.Number(value=0.0, label="Angle 6", step=5.0)
-# 		speed_angles = gr.Slider(value=50.0, minimum=0.0, maximum=100.0, step=1.0, label="Speed")
-
-# 	with gr.Group(visible=False) as control_coords_group:
-# 		coord1 = gr.Number(value=0.0, label="X coordinate", step=5.0)
-# 		coord2 = gr.Number(value=0.0, label="Y coordinate", step=5.0)
-# 		coord3 = gr.Number(value=0.0, label="Z coordinate", step=5.0)
-# 		coord4 = gr.Number(value=0.0, label="Rotation X", step=5.0)
-# 		coord5 = gr.Number(value=0.0, label="Rotation Y", step=5.0)
-# 		coord6 = gr.Number(value=0.0, label="Rotation Z", step=5.0)
-# 		speed_coords = gr.Slider(value=50.0, minimum=0.0, maximum=100.0, step=1.0, label="Speed")
-
-# 	with gr.Group(visible=False) as control_gripper_group:
-# 		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="Speed")
-
-# 	def toggle_visibility(command):
-# 		return (
-# 			gr.update(visible=command == "query/angles"),
-# 			gr.update(visible=command == "query/coords"),
-# 			gr.update(visible=command == "query/camera"),
-# 			gr.update(visible=command == "query/gripper"),
-# 			gr.update(visible=command == "control/angles"),
-# 			gr.update(visible=command == "control/coords"),
-# 			gr.update(visible=command == "control/gripper"),
-# 			gr.update(visible=False),
-# 			gr.update(value="")
-# 		)
-
-# 	submit_button = gr.Button("Send Command")
-# 	output_text = gr.Textbox(label="")
-# 	output_image = gr.Image(visible=False)
-
-# 	selected_command.change(
-# 		toggle_visibility,
-# 		selected_command,
-# 		[
-# 			query_angles_group, query_coords_group, query_camera_group, query_gripper_group,
-# 			control_angles_group, control_coords_group, control_gripper_group, output_image,
-# 			output_text
-# 		]
+# from collections import deque
+# import threading
+# import time
+
+# SESSION_TIME = 30
+
+# 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:
+# 				# Calculate remaining time for the current user
+# 				remaining_time = max(0, SESSION_TIME - (time.time() - self.session_start_time))
+# 				return 0, remaining_time
+# 			elif user_id in self.queue:
+# 				# Calculate wait time based on the queue position
+# 				position = list(self.queue).index(user_id) + 1
+# 				if self.session_start_time:
+# 					# Use the current session time to calculate wait time
+# 					wait_time = (position - 1) * SESSION_TIME + max(0, SESSION_TIME - (time.time() - self.session_start_time))
+# 				else:
+# 					# If no session is active, assume each user gets the full session time
+# 					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 process_input(input_text, 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:  # User is currently being served
+#         result = f"Processed: {input_text}"
+#         remaining_time = wait_time
+#         return result, f"Your turn! Time remaining: {remaining_time:.0f} seconds."
+#     elif position is not None:
+#         return None, f"You are in position {position}. Estimated wait time: {wait_time:.0f} seconds."
+#     else:
+#         return None, "Error: You are not in the queue."
+
+
+# def gradio_function(input_text, user_id):
+# 	result, status = process_input(input_text, user_id)
+# 	return result, status
+
+# def get_queue_status(user_id):
+# 	if not user_id:
+# 		return "Enter a User ID and submit to join the queue."
+# 	position, wait_time = queue_system.get_queue_info(user_id)
+# 	if position == 0:
+# 		return f"Your turn! Time remaining: {wait_time:.0f} seconds."
+# 	elif position is not None:
+# 		return f"You are in position {position}. Estimated wait time: {wait_time:.0f} seconds."
+# 	else:
+# 		return "You are not in the queue. Submit to join."
+
+# 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)
+
+# # Start the background timer thread
+# timer_thread = threading.Thread(target=background_timer, daemon=True)
+# timer_thread.start()
+
+# with gr.Blocks() as demo:
+# 	input_text = gr.Textbox(label="Input")
+# 	user_id = gr.Textbox(label="User ID")
+# 	output_text = gr.Textbox(label="Output")
+# 	status_text = gr.Markdown("Enter a User ID and submit to join the queue.")
+# 	submit_btn = gr.Button("Submit")
+
+# 	submit_btn.click(
+# 		fn=gradio_function,
+# 		inputs=[input_text, user_id],
+# 		outputs=[output_text, status_text]
 # 	)
 
-# 	submit_button.click(
-# 		handle_request,
-# 		inputs=[selected_command, angle1, angle2, angle3, angle4, angle5, angle6,
-# 				coord1, coord2, coord3, coord4, coord5, coord6, gripper_value, speed_angles,
-# 				speed_coords, speed_gripper],
-# 		outputs=[output_text, output_image],
-# 		concurrency_id="single_user",
-# 		concurrency_limit=1
+# 	demo.load(
+# 		fn=get_queue_status,
+# 		inputs=user_id,
+# 		outputs=status_text,
 # 	)
 
-# app.queue(default_concurrency_limit=1, max_size=100)
-# app.launch()
-import gradio as gr
-from collections import deque
-import threading
-import time
-
-SESSION_TIME = 30
-
-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
-                wait_time = (position - 1) * SESSION_TIME + max(0, SESSION_TIME - (time.time() - self.session_start_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 process_input(input_text, user_id):
-    update_queue()
-    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:  # User is currently being served
-        result = f"Processed: {input_text}"
-        remaining_time = wait_time
-        return result, f"Your turn! Time remaining: {remaining_time:.0f} seconds."
-    elif position is not None:
-        return None, f"You are in position {position}. Estimated wait time: {wait_time:.0f} seconds."
-    else:
-        return None, "Error: You are not in the queue."
-
-def update_queue():
-    if queue_system.end_session():
-        next_user = queue_system.dequeue_user()
-        if next_user:
-            queue_system.start_session(next_user)
-
-def gradio_function(input_text, user_id):
-    result, status = process_input(input_text, user_id)
-    return result, status
-
-def get_queue_status(user_id):
-    update_queue()
-    if not user_id:
-        return "Enter a User ID and submit to join the queue."
-    position, wait_time = queue_system.get_queue_info(user_id)
-    if position == 0:
-        return f"Your turn! Time remaining: {wait_time:.0f} seconds."
-    elif position is not None:
-        return f"You are in position {position}. Estimated wait time: {wait_time:.0f} seconds."
-    else:
-        return "You are not in the queue. Submit to join."
-
-with gr.Blocks() as demo:
-    input_text = gr.Textbox(label="Input")
-    user_id = gr.Textbox(label="User ID")
-    output_text = gr.Textbox(label="Output")
-    status_text = gr.Markdown("Enter a User ID and submit to join the queue.")
-    submit_btn = gr.Button("Submit")
-
-    submit_btn.click(
-        fn=gradio_function,
-        inputs=[input_text, user_id],
-        outputs=[output_text, status_text]
-    )
-
-    demo.load(
-        fn=get_queue_status,
-        inputs=user_id,
-        outputs=status_text,
-    )
-
-demo.queue(default_concurrency_limit=1, max_size=100)
-demo.launch()
+# demo.queue(default_concurrency_limit=1, max_size=100)
+# demo.launch()