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