기타 인터페이스

from interface.services import NotifyService, register_service

class NotifyService(NotifyService):
    service_type = "notify"
    service_name = "FANUC_M410IC_185_COMPACT"

    def __init__(self, update_success_num, update_fail_num):
        self.update_success_num = update_success_num
        self.update_fail_num = update_fail_num

    def handle_message(self, msg):
        if msg = "Success":
            self.update_success_num()
        elif msg = "Fail":
            self.update_fail_num()

pythonCopied!

class MyClient(TcpClientAdapter):

     def __init__(self, host_address):
         super().__init__(host_address)
         self._register_service()

     def _register_service(self):
         self.robot_service = NotifyService(self.update_success_num, self.update_fail_num)
         self.server, port = register_service(self.hub_caller, self.robot_service)

     def update_success_num(self):
         # the num of unstack successfully plus 1
         self.success_num += 1

     def update_fail_num(self):
         # the num of unstack fiplus 1
         self.fail_num += 1

pythonCopied!

from interface.services import VisionResultSelectedAtService, register_service

class VisionWatcher(VisionResultSelectedAtService):
    def __init__(self, send_err_no_cloud):
        super().__init__()
        self.send_err_no_cloud = send_err_no_cloud

    def poses_found(self,result):
        has_cloud_in_roi = not result.get("noCloudInRoi", False)

        if not has_cloud_in_roi:
            time.sleep(2)
            self.send_err_no_cloud()

pythonCopied!

class MyClient(TcpClientAdapter):
    def __init__(self, host_address):
        super().__init__(host_address)
        self._register_service()
    def _register_service(self):
        self.robot_service = VisionWatcher(self.send_err_no_cloud)
        self.server, port = register_service(self.hub_caller, self.robot_service)

    def send_err_no_cloud(self):
        # send no cloud error message
        self.send("12,NoCloudErr,done".encode())

pythonCopied!