Method predictor_worker

afy/predictor_worker.py:101–181

(recv_queue, send_queue, worker_alive)

    @staticmethod
    def predictor_worker(recv_queue, send_queue, worker_alive):
        predictor = None
        predictor_args = ()
        timing = AccumDict()
        log = Logger('./var/log/predictor_worker.log', verbose=opt.verbose)
        
        try:
            while worker_alive.value:
                tt = TicToc()

                try:
                    method, data = recv_queue.get(timeout=GET_TIMEOUT)
                except queue.Empty:
                    continue

                # get the latest non-critical request from the queue
                # don't skip critical request
                while not recv_queue.empty() and not method['critical']:
                    log(f"skip {method}")
                    method, data = recv_queue.get()

                log("working on", method)

                try:
                    tt.tic()
                    if method['name'] == 'predict':
                        image = cv2.imdecode(np.frombuffer(data, dtype='uint8'), -1)
                    else:
                        args = msgpack.unpackb(data)
                    timing.add('UNPACK', tt.toc())
                except ValueError:
                    log("Invalid Message", important=True)
                    continue

                tt.tic()
                if method['name'] == "hello":
                    result = "OK"
                elif method['name'] == "__init__":
                    if args == predictor_args:
                        log("Same config as before... reusing previous predictor")
                    else:
                        del predictor
                        predictor_args = args
                        predictor = PredictorLocal(*predictor_args[0], **predictor_args[1])
                        log("Initialized predictor with:", predictor_args, important=True)
                    result = True
                    tt.tic() # don't account for init
                elif method['name'] == 'predict':
                    assert predictor is not None, "Predictor was not initialized"
                    result = getattr(predictor, method['name'])(image)
                else:
                    assert predictor is not None, "Predictor was not initialized"
                    result = getattr(predictor, method['name'])(*args[0], **args[1])
                timing.add('CALL', tt.toc())

                tt.tic()
                if method['name'] == 'predict':
                    assert isinstance(result, np.ndarray), f'Expected np.ndarray, got {result.__class__}'