apartcore.py 文件源码

def __init__(self, listeners: List[MessageListener] = None,
                 on_finish: Callable[[int], None] = lambda return_code: None):
        """Starts an apart-core command and starts listening for zmq messages on this new thread"""
        Thread.__init__(self, name='apart-core-runner')
        self.ipc_address = 'ipc:///tmp/apart-gtk-{}.ipc'.format(uuid.uuid4())
        self.zmq_context = zmq.Context()
        self.socket = self.zmq_context.socket(zmq.PAIR)
        self.socket.setsockopt(zmq.RCVTIMEO, 100)
        self.socket.bind(self.ipc_address)
        self.on_finish = on_finish
        self.listeners = listeners or []  # List[MessageListener]

        if LOG_MESSAGES:
            self.register(MessageListener(lambda msg: print('apart-core ->\n {}'.format(str(msg)))))

        # Current default is apart-core binary stored in the directory above these sources
        apart_core_cmd = os.environ.get('APART_GTK_CORE_CMD') or \
            os.path.abspath(os.path.dirname(os.path.realpath(__file__)) + '/../apart-core')

        try:
            if os.geteuid() == 0 or os.environ.get('APART_PARTCLONE_CMD'):
                self.process = subprocess.Popen([apart_core_cmd, self.ipc_address])
            else:
                self.process = subprocess.Popen(['pkexec', apart_core_cmd, self.ipc_address])
        except FileNotFoundError:
            if os.geteuid() == 0:
                print('apart-core command not found at \'' + apart_core_cmd + '\'', file=sys.stderr)
            else:
                print('pkexec command not found, install polkit or run as root', file=sys.stderr)
            self.zmq_context.destroy()
            sys.exit(1)
        self.start()