Abstract Parent Class Interfaces¶
Abstract Parent Class Interfaces are the functions that can be added with a child class that will inherit from its parent class. You can rewrite the functions according to actual requirements. This section will introduce you to the following abstract parent classes.
Communication¶
Source code belonging to the communication category is stored in the /src/interface/communication.py
file in the installation directory of Mech-Center.
Communication Class¶
Communication class is a basic class used for communication. It provides a series of interface functions that should be rewritten on the server or client.
Class Function |
Description |
is_connected() |
Determine if the current connection is successful. |
set_recv_size() |
Set the length of the data received each time; the default value is 1024 bytes. |
send() |
Interface function which is used to send data. |
recv() |
Interface function which is used to receive data. |
close() |
Interface function which is used to close the socket communication. |
before_recv() |
Interface function which you can add a logic according to actual needs before receiving the data. Function overriding is available. |
after_recv() |
Interface function which you can add a logic according to actual needs after receiving the data. Function overriding is available. |
after_handle() |
Interface function which you can add a logic according to actual needs after processing the data. Function overriding is available. |
TcpServer Class¶
TcpServer class encapsulates a TCP/IP Socket server.
Class Function |
Description |
bind_and_listen() |
Bind the port |
local_socket() |
Provide the local Socket information |
remote_socket() |
Provide the remote Socket information |
accept() |
Accept the connection with the client |
send() |
Send the data |
recv() |
Receive the data |
close() |
Close the socket |
close_client() |
Disconnect with the client |
TcpClinet Class¶
TcpClient class encapsulates a TCP/IP Socket client.
Class Property |
Description |
is_bind_port |
Whether to bind the port. If there is a port restriction on the client, the value should be set to True. |
Class Function |
Description |
send() |
Send the data |
recv() |
Receive the data |
close() |
Close the socket |
set_timeout() |
Timeout period in seconds |
reconnect_server() |
Reconnect to the server |
after_connect_server() |
Interface function which is used to specify what to do after the first successful connection with the server |
after_reconnect_server() |
Interface function which is used to specify what to do after reconnecting with the server |
after_timeout() |
Interface function which is used to specify what to do after the timeout period |
Adapter¶
Source code belonging to the Adapter category is stored in the /src/interface/adapter.py
file in the installation directory of Mech-Center.
Adapter Base Class¶
Adapter encapsulates functions related to Mech-Viz, Mech-Vision, Mech-Center, Robserver, including start_viz(), stop_viz(), set_task_property(), and set_step_property(). When an Adapter program is used to control Mech-Viz or Mech-Vision, the child class of Adapter base class must be used.
The class properties of Adapter are shown in the table below.
Class Property |
Description |
viz_project_dir |
Directory of the current Mech-Viz project |
vision_project_name |
Name of the current Mech-Vision project |
is_simulate |
Whether to run Mech-Viz in simulation mode |
is_keep_viz_state |
Whether to keep the state when Mech-Viz stopped last time |
is_save_executor_data |
Whether to save the data of Mech-Viz executor |
is_force_simulate |
Whether to force Mech-Viz to run in simulation mode |
is_force_real_run |
Whether to force Mech-Viz to run and control the real robot |
code_signal |
The signal of displaying Adapter information in the main interface of Mech-Center (including error codes) |
msg_signal |
The signal of displaying Adapter information in the main interface of Mech-Center (NOT including error codes) |
i_code_signal |
The signal of displaying Mech-Interface information in the main interface of Mech-Center (including error codes) |
i_msg_signal |
The signal of displaying Mech-Interface information in the main interface of Mech-Center (NOT including error codes) |
viz_finished_signal |
The signal indicating that Mech-Viz is stopped normally or with an error |
connect_robot_signal |
Connect/disconnect with the robot signal |
start_adapter_signal |
Start Adapter signal |
service_name_changed |
The signal of displaying statuses of Mech-Viz and Mech-Vision in the main interface of Mech-Center |
setting_infos |
The configuration information of Mech-Center |
service_name |
Name of the registered service |
The Adapter class functions are as shown in the table below.
Class Function |
Description |
on_exec_status_changed() |
Receive the status information from Mech-Viz and Mech-Vision |
register_self_service() |
Register Adapter service |
vision_project_dirs(self): |
Check the directory of Mech-Vision project |
vision_project_names() |
Check all project names in Mech-Vision |
vision_project_names_in_center() |
Check the names of all Mech-Vision projects loaded in Mech-Center |
is_viz_registered() |
Check whether Mech-Viz is registered in Mech-Center |
is_viz_in_running() |
Check whether Mech-Viz is running |
is_vision_started() |
Check whether Mech-Vision project is registered in Mech-Center |
find_services() |
Check the service |
before_start_viz() |
This function will be called before starting Mech-Viz |
after_start_viz() |
This function will be called after starting Mech-Viz |
viz_not_registerd() |
This function will be called if Mech-Viz is not registered after being started |
viz_is_running() |
This function will be called if Mech-Viz is already running when starting Mech-Viz |
viz_run_error() |
This function will be called if an error occurs when running Mech-Viz |
viz_run_finished() |
This function will be called after Mech-Viz is stopped |
viz_plan_failed() |
This function will be called after Mech-Viz fails to plan a path |
viz_unreachable_targets() |
This function will be called when there is an unreachable target in the path planned by Mech-Viz |
viz_collision_checked() |
This function will be called when a collision is detected in the path planned by Mech-Viz |
parse_viz_reply() |
Parse the reply from Mech-Viz |
wait_viz_result() |
Wait for Mech-Viz to reply |
start_viz() |
Start Mech-Viz |
stop_viz() |
Stop Mech-Viz |
pause_viz() |
Pause Mech-Viz |
find_vision_pose() |
Trigger Mech-Vision project to capture images |
async_call_vision_run() |
Trigger Mech-Vision project to capture images asynchronously |
async_get_vision_callback() |
Receive the result from Mech-Vision asynchronously |
deal_vision_result() |
Process the result from Mech-Vision |
set_step_property() |
Set Step parameters in Mech-Vision |
read_step_property() |
Read Step parameters in Mech-Vision |
select_parameter_group() |
Select the parameter recipe in Mech-Vision |
set_task_property() |
Set Step parameters in Mech-Viz |
read_task_property() |
Read Step parameters in Mech-Viz |
get_digital_in() |
Obtain DI |
set_digital_out() |
Set DO |
before_start_adapter() |
This function will be called before starting Adapter |
start() |
Start Adapter |
close() |
Stop Adapter |
handle_command() |
Process commands received from external devices |
TcpServerAdapter Class¶
As shown below, TcpServerAdapter class inherits from Adapter, and it encapsulates functions of TcpServer.
class TcpServerAdapter(Adapter):
def __init__(self, host_address, server=TcpServer):
super(TcpServerAdapter, self).__init__()
self.init_server(host_address, server)
def init_server(self, host_address, server=TcpServer):
self._server = server(host_address)
def set_recv_size(self, size):
self._server.set_recv_size(size)
def send(self, msg, is_logging=True):
return self._server.send(msg, is_logging)
def recv(self):
return self._server.recv()
def start(self):
self.before_start_adapter()
while not self.is_stop_adapter:
try:
self._server.before_recv()
cmds = self._server.recv()
logging.info("Received raw data from client:{}".format(cmds))
if not cmds:
logging.warning("Adapter client is disconnected!")
self.code_signal.emit(logging.WARNING, CENTER_CLIENT_DISCONNECTED)
self._server.close_client()
self.accept()
continue
self._server.after_recv()
except socket.error:
logging.warning("Adapter client is closed!")
self.code_signal.emit(logging.WARNING, CENTER_CLIENT_DISCONNECTED)
self._server.close_client()
self.accept()
except Exception as e:
logging.exception("Exception occurred when receiving data from client: {}.".format(e))
else:
try:
self.handle_command(cmds)
self._server.after_handle()
except Exception as e:
self.msg_signal.emit(logging.ERROR, _translate("messages", "Handle command exception: {}".format(e)))
logging.exception("Adapter exception in handle_command(): {}".format(e))
def close(self):
super().close()
self._server.close()
def before_start_adapter(self):
super().before_start_adapter()
self.accept()
def accept(self):
if self.is_stop_adapter:
return
self.code_signal.emit(logging.INFO, CENTER_WAIT_FOR_CLIENT)
self._server.accept()
if self._server.is_connected():
self.code_signal.emit(logging.INFO, CENTER_CLIENT_CONNECTED)
self.msg_signal.emit(logging.INFO, _translate("messages", "Client address is") + " {}".format(self._server.remote_socket()[1]))
TcpClientAdapter Class¶
As shown below, TcpClientAdapter class inherits from Adapter, and it encapsulates functions of TcpClient.
class TcpClientAdapter(Adapter):
def __init__(self, host_address):
super().__init__()
self.init_client(host_address)
def init_client(self, host_address, client=TcpClient):
self._client = client(host_address)
def set_bind_port(self, is_bind=True):
self._client.is_bind_port = is_bind
def set_recv_size(self, size):
self._client.set_recv_size(size)
def send(self, msg, is_logging=True):
self._client.send(msg, is_logging)
def recv(self):
return self._client.recv()
def start(self):
self.reconnect_server(False)
while not self.is_stop_adapter:
try:
self._client.before_recv()
cmds = self._client.recv()
if not cmds:
self.reconnect_server()
continue
logging.info("Received command from server:{}".format(cmds))
self._client.after_recv()
except socket.timeout:
logging.warning("Socket timeout")
self._client.after_timeout()
except socket.error:
sleep(5)
self.reconnect_server()
except Exception as e:
logging.exception("Exception occurred when receiving from server: {}".format(e))
else:
try:
self.handle_command(cmds)
except Exception as e:
self.msg_signal.emit(logging.ERROR, _translate("messages", "Handle command exception: {}".format(e)))
logging.exception("Adapter exception in handle_command(): {}".format(e))
def close(self):
super().close()
self._client.close()
def reconnect_server(self, is_reconnect=True):
self._client.reconnect_server()
if self.is_stop_adapter:
return
if self._client.is_connected():
self.code_signal.emit(logging.INFO, CENTER_CONNECT_TO_SERVER)
else:
self.code_signal.emit(logging.WARNING, CENTER_SERVER_DISCONNECTED)
TcpMultiplexingServerAdapter Class¶
As shown below, TcpMultiplexingServerAdapter class inherits from Adapter, and it is mainly used to connect multiple clients.
class TcpMultiThreadingServerAdapter(Adapter):
def __init__(self, address):
super().__init__()
self._servers = {}
self.add_server(address)
self.sockets = {}
self.clients_ip = {}
self.thread_pool = ThreadPoolExecutor(max_workers=4, thread_name_prefix="tcp_multi_server_thread")
self.thread_id_socket_dict = {}
self.set_recv_size()
def set_recv_size(self, size=1024):
self.recv_size = size
def _find_client_ip(self, sock):
for k, v in self.sockets.items():
if v == sock:
return k
def _find_server(self, sock):
for k, v in self._servers.items():
if v == sock:
return k
def add_server(self, host_address):
server = TcpServer(host_address)
server.bind_and_listen()
self._servers[server] = server.local_socket()
def set_clients_ip(self, clients_ip):
"""
Must be called before start().
`clients_ip` is a dict(key is client ip, value is client description).
"""
self.clients_ip = clients_ip
def add_connection(self, ip_port, sock):
self.sockets[ip_port] = sock
logging.info("Add {}, connections: {}".format(ip_port, self.sockets))
self.msg_signal.emit(logging.INFO, _translate("messages", "The client {} gets online.").format(ip_port))
def del_connection(self, ip):
logging.info("Del {}, connections: {}".format(ip, self.sockets))
if self.client_connection(ip):
self.client_connection(ip).close()
self.sockets.pop(ip)
self.msg_signal.emit(logging.WARNING, _translate("messages", "The client {} gets offline.").format(ip))
def client_connection(self, client_ip):
return self.sockets.get(client_ip)
def check_read_events(self, rs):
for s in rs:
if s in self._servers.values(): # recv connection
server = self._find_server(s)
if self.is_stop_adapter:
return
server.accept()
client_socket, client_addr = server.remote_socket()
ip_port = "{}:{}".format(str(client_addr[0]), str(client_addr[1]))
self.add_connection(ip_port, client_socket)
elif s in self.sockets.values(): # recv data
client_ip = self._find_client_ip(s)
if not client_ip:
continue
msg = self.recv_by_s(s)
if not msg:
self.del_connection(client_ip)
return
try:
future = self.thread_pool.submit(self.handle_command_thread, s, msg)
except Exception as e:
logging.exception("Adapter exception in handle_command(): {}".format(e))
def handle_command_thread(self, s, msg):
thread_id = threading.get_ident()
self.thread_id_socket_dict[thread_id] = s
self.handle_command(msg)
# del self.thread_id_socket_dict[thread_id]
def send(self, msg, is_logging=True):
thread_id = threading.get_ident()
sock = self.thread_id_socket_dict.get(thread_id)
len_total = len(msg)
while msg:
if sock:
len_sent = sock.send(msg)
else:
for v in self.sockets.values():
try:
len_sent = v.send(msg)
except Exception as e:
logging.warning(e)
if not len_sent:
logging.warning("Connection lost, close the client connection.")
return len_sent
if is_logging:
logging.info("Server send: {}, len_sent: {}".format(msg, len_sent))
msg = msg[len_sent:]
return len_total
def recv(self):
thread_id = threading.get_ident()
sock = self.thread_id_socket_dict.get(thread_id)
return self.recv_by_s(sock)
def recv_by_s(self, sock):
msg = b""
try:
msg = sock.recv(self.recv_size)
except socket.error:
logging.error("The client is closed!")
if msg:
logging.info("Received message: {}".format(msg))
return msg
def check_task(self):
"""
Interface.
"""
def close(self):
super().close()
for server in self._servers.keys():
server.close()
for client_ip in self.sockets.keys():
try:
self.client_connection(client_ip).close()
logging.info("Close socket :{}".format(client_ip))
except Exception as e:
logging.warning("Close socket error:{}, exception:{}".format(client_ip, e))
self.sockets = {}
def start(self):
self.before_start_adapter()
while not self.is_stop_adapter:
avalible_sockets = list(self.sockets.values()) + list(self._servers.values())
rs, _, _ = select(avalible_sockets, [], [], 0.1)
self.check_read_events(rs)
try:
self.check_task()
except Exception as e:
self.msg_signal.emit(logging.ERROR,
_translate("messages", "Handle command exception: {}".format(e)))
logging.exception("Exception when check task:{}".format(e))
sleep(5)
IOAdapter Class¶
As shown below, IOAdapter class inherits from Adapter, and it uses a while loop in obtaining DI.
class IOAdapter(Adapter):
robot_name = None
check_rate = 0.5
def __init__(self, host_address):
super().__init__()
self.last_gi = 0
def get_digital_in(self, timeout=None):
return super().get_digital_in(self.robot_name, timeout)
def set_digital_out(self, port, value, timeout=None):
super().set_digital_out(self.robot_name, port, value, timeout)
def _check_gi(self):
gi_js = self.get_digital_in()
gi = int(json.loads(gi_js.decode())["value"])
if self.last_gi != gi:
self.last_gi = gi
logging.info("Check GI signal status: {}".format(gi))
self.handle_gi(gi)
def start(self):
self.before_start_adapter()
while not self.is_stop_adapter:
try:
self._check_gi()
except Exception as e:
logging.exception(e)
self.check_gi_failed()
sleep(self.check_rate)
def handle_gi(self, gi):
"""
Interface.
"""
def check_gi_failed(self):
"""
Interface.
"""
AdapterWidget Class¶
As shown below, AdapterWidget class is a parent class, and all functions related to user interface customization must be its child classes.
class AdapterWidget(QWidget):
def set_adapter(self, adapter):
self.adapter = adapter
self.after_set_adapter()
def after_set_adapter(self):
"""
Interface.
"""
def close(self):
super().close()
"""
Interface.
"""
Service¶
Source code belonging to the service category is stored in the /src/interface/services.py
file in the installation directory of Mech-Center.
NotifyService Class¶
The code of NotifyService class is shown below.
class NotifyService(JsonService):
service_type = "notify"
service_name = "adapter"
def handle_message(self, msg):
"""
Interface.
"""
def notify(self, request, _):
msg = request["notify_message"]
logging.info("notify message:{}".format(msg))
return self.handle_message(msg)
The default service name is “adapter”. If multiple notify services are needed in the project, you can override the service_name to distinguish different services.
The descriptions of class functions are shown in the table below.
Class Function |
Description |
handle_message() |
Interface function; the child class can be overridden |
notify() |
Parse the message; the child class usually does not need to be overridden |
VisionResultSelectedAtService Class¶
The code of VisionResultSelectedAtService class is shown below.
class VisionResultSelectedAtService(JsonService):
service_type = "vision_watcher"
service_name = "vision_watcher_adapter"
def __init__(self):
self.poses = None
def poses_found(self, result):
"""
Interface.
"""
def posesFound(self, request, _):
logging.info("{} result:{}".format(jk.mech_vision, request))
self.poses_found(request)
def poses_planned(self, result):
"""
Interface.
"""
def posesPlanned(self, request, _):
logging.info("Plan result:{}".format(request))
self.poses_planned(request)
def multiPickCombination(self, request, _):
logging.info("multiPickCombination:{}".format(request))
The default service type is vision_watcher, which cannot be modified. The default service name is vision_watcher_adapter. If multiple vision_watcher services are needed in the project, you can modify the service_name in the child class to distinguish different services.
The descriptions of class functions are shown in the table below.
Class Function |
Description |
poses_found() |
Interface function whose child class can be overridden; the parameter is the recognition result of Mech-Vision |
posesFound() |
Parse the recognition result sent from Mech-Vision; child class function usually does not need to be overridden |
poses_planned() |
Interface function; the parameter is the vision point in the path planned by Mech-Viz |
posesPlanned() |
Parse the planning message sent from Mech-Viz |
RobotService Class¶
The RobotService class is shown below.
class RobotService(JsonService):
service_type = "robot"
service_name = "robot"
jps = [0, 0, 0, 0, 0, 0]
pose = [0, 0, 0, 1, 0, 0, 0]
def getJ(self, *_):
return {"joint_positions": self.jps}
def setJ(self, jps):
logging.info("setJ:{}".format(jps))
self.jps = jps
def getL(self, *_):
return {"tcp_pose": self.pose}
def getFL(self, *_):
return {"flange_pose": self.pose}
def setL(self, pose):
logging.info("setL:{}".format(pose))
self.pose = pose
def moveXs(self, params, _):
pass
def stop(self, *_):
pass
def setTcp(self, *_):
pass
def setDigitalOut(self, params, _):
pass
def getDigitalIn(self, *_):
pass
def switchPauseContinue(self, *_):
pass
The default service type is robot, which cannot be modified. The default service name is robot, which should be modified into the name of the real robot in the child class. You will need to set a home position in JPs or TCP in the child class. Please make sure that this pose will not cause collision with the scene objects.
The descriptions of class functions are shown in the table below.
Class Function |
Description |
getJ() |
Return JPs to Mech-Viz/Mech-Vision |
setJ() |
Set JPs in radians for external srvices |
getL() |
Return TCP to Mech-Viz/Mech-Vision |
getFL() |
Return flange pose to Mech-Viz/Mech-Vision |
setL() |
Set flange pose in quaternions for external srvices; the unit is meters |
moveXs() |
This function will be called after Mech-Viz finishes planning the path. The parameter contains property of targets from move Steps. Please note that if Steps that may interrupt pre-planning, such as Check DI, Branch by Msg, Mech-Viz will call this function for multiple times. |
stop() |
Stop the robot (this function is seldom called) |
setTcp() |
Set TCP (this function is seldom called) |
setDigitalOut() |
Set DO (this function is seldom called) |
getDigitalIn() |
Obtain DI (this function is seldom called) |
switchPauseContinue() |
Pause/continue the robot (this function is seldom called) |
OuterMoveService Class¶
The code of OuterMoveService class is shown below.
class OuterMoveService(JsonService):
service_type = "outer_move"
service_name = "outer_move"
move_target_type = TCP_POSE
velocity = 0.25
acceleration = 0.25
blend_radius = 0.05
motion_type = MOVEJ
is_tcp_pose = False
pick_or_place = 0
def __init__(self):
self.targets = []
def gather_targets(self, di, jps, flange_pose):
"""
Interface.
Please add targets to `self.targets` here if needed.
"""
def add_target(self, move_target_type, target):
self.targets.append({"move_target_type": move_target_type, "target": target})
def getMoveTargets(self, params, *_):
"""
@return: targets(move_target_type 0:jps, 1:tcp_pose, 2:obj_pose)
velocity(default 0.25)
acceleration(default 0.25)
blend_radius(default 0.05)
motion_type(default moveJ 'J':moveJ, 'L':moveL)
is_tcp_pose(default False)
"""
di = params["di"]
jps = params["joint_positions"]
flange_pose = params["pose"]
logging.info("getMoveTargets: di={}, jps={}, flange_pose={}".format(di, jps, flange_pose))
self.gather_targets(di, jps, flange_pose)
targets = self.targets[:]
self.targets.clear()
logging.info("Targets: {}".format(targets))
return {"targets": targets, "velocity": self.velocity, "acceleration": self.acceleration, "blend_radius": self.blend_radius,
"motion_type": self.motion_type, "is_tcp_pose": self.is_tcp_pose, "pick_or_place": self.pick_or_place}
Both the default service type and service name are External Move. If multiple External Move services are needed in the project, you can modify the service_name in the child class to distinguish different services.
The descriptions of class functions are shown in the table below.
Class Function |
Description |
move_target_type() |
Type of the target; 0:jps, 1:tcp_pose, 2:obj_pose. |
velocity() |
Speed of the target; the default value is 0.25. |
acceleration() |
Acceleration of the target; the default value is 0.25. |
blend_radius() |
The turning radius of the target; the default value is 0.05m. |
motion_type() |
The motion type of the target: ‘J’:moveJ, ‘L’:moveL. |
is_tcp_pose() |
Whether the pose of the target is in TCP. |
gather_targets() |
Interface function, which collects all the targets. The parameter contains current JPs, flange pose, and DI value of the robot. The child class can be overriden according to actual requirements. |
add_target() |
Add a target; this function can be called in the child class to add a target. |
getMoveTargets() |
This function will be called when Mech-Viz proceeds on the External Move Step. The parameter contains current JPs, flange pose, and DI value of the robot. |
Register Service¶
The services corresponding to the above 4 classes are only available after being registered. The function used for service registration is shown below.
def register_service(hub_caller, service, other_info=None):
server, port = start_server(service)
if service.service_type == "robot":
other_info["from_adapter"] = True
other_info["simulate"] = False
hub_caller.register_service(service.service_type, service.service_name, port, other_info)
return server, port