asyncio와 Kivy 믹싱 : asyncio 루프와 Kivy 애플리케이션을 동시에 시작하는 방법은 무엇입니까?

asyncio에서 손실됩니다.asyncio와 Kivy 믹싱 : asyncio 루프와 Kivy 애플리케이션을 동시에 시작하는 방법은 무엇입니까?

저는 Kivy와 asyncio를 동시에 배우고 있습니다. Kivy를 실행하고 asyncio 루프를 실행하는 문제를 해결할 수있었습니다. 어떻게 전환했는지는 모두 호출을 차단하고 실행해야합니다. 순차적으로 (잘, 내가 틀렸 으면 좋겠다.) 예.

loop = asyncio.get_event_loop() 
loop.call_soon(MyAsyncApp().run()) 
loop.run_forever()

현재 시도로 인해 응용 프로그램이 시작되지만 coroutine이 실행되지 않습니다. "Connect (연결)"버튼을 클릭하면 loop.call_soon을 사용하여 작업을 예약하고 실행해야하지만 아무 일도 일어나지 않습니다.

누군가 내 코드를보고 문제에 대한 적절한 접근을 제안 할 수 있습니까? 응용 프로그램의

import asyncio 
import random 
import time 
from kivy.app import App 
from kivy.lang import Builder 

ui = Builder.load_string(''' 
BoxLayout: 
    orientation: 'vertical' 
    GridLayout: 
     rows: 2 
     cols: 2 
     Label: 
      text: 'Status:' 
      size_hint: 0.3, 1 
     Label: 
      id: status 
      text: '' 
     Label: 
      text: 'Data:' 
      size_hint: 0.7, 1 
     Label: 
      id: data 
      text: '' 
    BoxLayout: 
     direction: 'horizontal' 
     Button: 
      text: 'Get Data' 
      on_press: app.connect() 
     Button: 
      text: 'Stop Data' 
      on_press: pass 
''') 

class MyAsyncApp(App): 

    def __init__(self): 
     super(self.__class__, self).__init__() 

     self.x_connected = None 
     self.x_widget_data = None 
     self.x_widget_status = None 
     self.x_loop = asyncio.get_event_loop() 

    def build(self): 
     return ui 

    def connect(self): 
     # Get widget 
     self.x_widget_status = self.root.ids.status 

     # Update status 
     self.x_widget_status.text = 'Preparing to connect...' 

     # Connect using asyncio 
     # --> But the loop must be already running <--- 
     self.x_loop.call_soon(self.do_connect) 

    async def do_connect(self): 
     # Connect asynchronously 

     # Get widget 
     self.x_widget_data = self.root.ids.data 

     # Update status 
     self.x_connected = False 
     self.x_widget_status.text = 'Connecting...' 

     # Perform actual actions 
     try: 
      result = await self.feed_kivy() 
      if result: 
       self.x_widget_status.text = 'Service not available: ' + result 
       return 
     except Exception as e: 
      self.x_widget_status.text = 'Error while connecting' 
      return 

     # Update status 
     self.x_connected = True 
     self.x_widget_status.text = 'Connected' 

    async def feed_kivy(self): 
     # Deliver fresh data at random interval 

     # Some slow process to get data 
     result = await asyncio.sleep(random.randint(1, 5), time.time()) 
     self.x_widget_data.text = result 

     # Reschedule ourselves 
     await self.x_loop.call_soon(self.feed_kivy()) 


def main(): 
    # If loop started here, app is never started 
    loop = asyncio.get_event_loop() 
    loop.call_soon(MyAsyncApp().run()) 
    loop.run_forever() 
    loop.close() 


if __name__ == '__main__': 
    main()

출처

2017-12-14 mins