Как постоянно обновлять интерфейс Kivy с помощью ответа API?

Question

Я пытаюсь создать базовую c систему, которая может взять кадр с веб-камеры и отправить его в Azure Face API для распознавания эмоций и показать результат на интерфейсе Kivy на моей ма c.

Я пытался:

1 - писать классы kivy mytest и MyApp внутри, пока l oop, но интерфейс завис из-за внутреннего l oop kivy.

2 - вызов функции result_print внутри l oop, например:

mytest().result_print(max_emotion, max_value) 

, но я получил ошибку, что класс не определен.

3 - попытка функция часов от Kivy, но после прочтения документации и предыдущего вопроса я не понял, где написать эту функцию. Я попытался написать его внутри l oop, но также получил ошибку, что функция не определена.

кстати, это мой первый вопрос о StackOverflow, и я только начал изучать OOP и киви 3 дня go.

Это мои блоки полного кода и надеюсь найти кого-нибудь может помочь как их скомпилировать.

#import ...    

# subscription key value
    subscription_key = "12345"
    assert subscription_key

    # endpoint URL
    face_api_url = 'https://emotion-recognition-project.cognitiveservices.azure.com/face/v1.0/detect'

    # headers
    headers = {'Ocp-Apim-Subscription-Key': subscription_key, 'Content-Type': 'application/octet-stream'}

    # parameters
    params = {
        'returnFaceId': 'true',
        'returnFaceLandmarks': 'false',
        'returnFaceAttributes': 'emotion',
    }
    # Webcam
    cap = WebcamVideoStream(src=0).start()
    count = 0

    while True:

        # read the frame
        frame = cap.read()

        # convert to gray
        gray =cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

        # write the image
        cv2.imwrite(os.path.join('REST_API_test/images', "frame{:d}.jpg".format(count)), gray)  # save frame as JPEG file

        image_path = "REST_API_test/images/frame{:d}.jpg".format(count)
        image_data = open(image_path, "rb").read()

        # send the image to API
        response = requests.post(face_api_url, params=params, headers=headers, data = image_data )
        # response = requests.post(face_api_url, params=params, headers=headers, json={"url": image_url})
        result = response.json()
        if len(result) > 0:
            print(len(result))

            # find the max emotion with max value
            emotion = result[0]["faceAttributes"]["emotion"]
            max_value = max(emotion.values())
            max_emotion = max(emotion, key = emotion.get)
            print("The emotion is",max_emotion, "{:.0%}".format(max_value))
            Clock.schedule_interval(partial(result_print, max_emotion ,max_value), 1)


        # remove the frame
        os.remove("REST_API_test/images/frame{:d}.jpg".format(count))
        count += 1



    class mytest(GridLayout):
        def __init__(self, **kwargs):
            super(mytest, self).__init__(**kwargs)

            self.key = max_emotion
            self.value = max_value

        def result_print(self, key, value):
            key = self.key
            value = self.value
            self.add_widget(Label(text = key, pos = (400,200)))
            self.add_widget(Label(text = "{:.0%}".format(value), pos = (700,200)))


    class MyApp(App):
      def build(self):
          return mytest()

    MyApp().run()

Answer 1

Я думаю, вы должны поместить код, который обращается к веб-камере, в отдельный поток, а затем вызвать Clock.schedule_once(), чтобы выполнить обновление интерфейса Kivy. Примерно так:

#import ...
import threading


class mytest(GridLayout):
        def __init__(self, **kwargs):
            super(mytest, self).__init__(**kwargs)

            self.key = max_emotion
            self.value = max_value

        # need a `dt` argument (added by Clock.schedule_once)
        def result_print(self, key, value, dt):
            key = self.key
            value = self.value
            self.add_widget(Label(text = key, pos = (400,200)))
            self.add_widget(Label(text = "{:.0%}".format(value), pos = (700,200)))

        def do_updates(self):
            # subscription key value
            subscription_key = "12345"
            assert subscription_key

            # endpoint URL
            face_api_url = 'https://emotion-recognition-project.cognitiveservices.azure.com/face/v1.0/detect'

            # headers
            headers = {'Ocp-Apim-Subscription-Key': subscription_key, 'Content-Type': 'application/octet-stream'}

            # parameters
            params = {
                'returnFaceId': 'true',
                'returnFaceLandmarks': 'false',
                'returnFaceAttributes': 'emotion',
            }
            # Webcam
            cap = WebcamVideoStream(src=0).start()
            count = 0

            while True:

                # read the frame
                frame = cap.read()

                # convert to gray
                gray =cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

                # write the image
                cv2.imwrite(os.path.join('REST_API_test/images', "frame{:d}.jpg".format(count)), gray)  # save frame as JPEG file

                image_path = "REST_API_test/images/frame{:d}.jpg".format(count)
                image_data = open(image_path, "rb").read()

                # send the image to API
                response = requests.post(face_api_url, params=params, headers=headers, data = image_data )
                # response = requests.post(face_api_url, params=params, headers=headers, json={"url": image_url})
                result = response.json()
                if len(result) > 0:
                    print(len(result))

                    # find the max emotion with max value
                    emotion = result[0]["faceAttributes"]["emotion"]
                    max_value = max(emotion.values())
                    max_emotion = max(emotion, key = emotion.get)
                    print("The emotion is",max_emotion, "{:.0%}".format(max_value))

                    # schedule update for next frame (leave out the timeout parameter)
                    Clock.schedule_interval(partial(self.result_print, max_emotion ,max_value))


                # remove the frame
                os.remove("REST_API_test/images/frame{:d}.jpg".format(count))
                count += 1


    class MyApp(App):
      def build(self):
          theRoot = mytest()

          # start the update thread
          threading.Thread(target=theRoot.do_updates, daemon=True).start()

          return theRoot

    MyApp().run()

Я не проверял это, но я думаю, что этот подход должен работать.