비동기인풋, toolbar 터미널출력 << prompt-toolkit, curses

코루틴 동작은 동작대로 하고

input을 비동기적으로 받는다.

받았을 시 동작 하게끔 가능 + input 메세지는 항상 맨 하단에 있다.

 

---

pip 로 패키지 받고

pip install -U prompt-toolkit

 

 

---

코드

import asyncio
from prompt_toolkit import PromptSession
from prompt_toolkit.patch_stdout import patch_stdout


class AA():
  def make_loop(self):
    self.main_loop = asyncio.get_event_loop()
    # self.main_loop.add_signal_handler(signal.SIGINT, self.ask_exit)
    # self.main_loop.add_signal_handler(signal.SIGTERM, self.ask_exit)
    self.main_loop.run_until_complete(self.make_task())
    # self.main_loop.run_forever()

  async def make_task(self):
    # ros2_task = asyncio.create_task(self.execute_spin())
    # mqtt_task = asyncio.create_task(self.mqtt_agent_main())
    await asyncio.gather(
      self.ros_task_main(),
      self.another_task()
    )
    print('TASK END ERROR ', flush=True)

  async def another_task(self):
    while True:
      await asyncio.sleep(1)

  async def ros_task_main(self):
    asyncio.create_task(self.publish_timer())
    asyncio.create_task(self.checking_input())

  async def checking_input(self):
    session = PromptSession()
    while True:
      with patch_stdout():
        input_command = await session.prompt_async('command 1(tts),2(expression) : ')
        if input_command == '1':
          tts_comment = await session.prompt_async('tts : ')
        elif input_command == '2':
          expression_number = await session.prompt_async('expression 예시: EX_01 ~ EX_10 : ')
        else:
          print('1,2 둘중에 하나만 고르숑')

      if input_command == '1':
        _msg = {'a': 'b', 'c': 'd', 'tts': tts_comment}
        print(_msg)
      elif input_command == '2':
        _msg = {'a': 'b', 'c': 'd', 'tts': expression_number}
        print(_msg)

  async def publish_timer(self):
    i = 0
    while True:
      print('counut : {}'.format(i))
      i = i + 1
      await asyncio.sleep(1)

aa=AA()
aa.make_loop()

 

출력결과 : 

 

 

---

코드 :

lib : curses_4

key val 타입으로 입력받아서, 하단의 toolbar 형식으로 출력

터미널세션을 나눠서 출력시키는 형

import asyncio
import curses

SPARE_B = 5
async def update_upper_terminal(stdscr):
    max_height, _ = stdscr.getmaxyx()
    upper_height = max_height - SPARE_B  # 하단 모니터링 바를 위해 한 줄을 남김

    while True:
        for i in range(upper_height):
            stdscr.move(i, 0)  # 커서를 상단 터미널의 시작점으로 이동
            stdscr.clrtoeol()  # 현재 라인을 지움
            stdscr.addstr(i, 0, "11111\n")  # 상단 터미널에 "11111" 출력
            await asyncio.sleep(0.1)  # 0.1초 대기
        await asyncio.sleep(0.1)  # 0.1초 대기

def draw_monitoring_bar(stdscr, data, width, height):
    # 색상 쌍 초기화
    curses.start_color()
    curses.init_pair(1, curses.COLOR_BLACK, curses.COLOR_GREEN)  # 검은색 글자, 초록색 배경
    curses.init_pair(2, curses.COLOR_WHITE, curses.COLOR_BLACK)  # 흰색 글자, 검은색 배경

    # 하단 모니터링 바 설정
    stdscr.attron(curses.color_pair(1))  # 키에 대한 색상 설정
    for idx, (key, value) in enumerate(data.items()):
        stdscr.addstr(height - SPARE_B, idx * 20, f" {key:<10}", curses.color_pair(1))
    stdscr.attroff(curses.color_pair(1))  # 키에 대한 색상 해제

    stdscr.attron(curses.color_pair(2))  # 값에 대한 색상 설정
    for idx, (key, value) in enumerate(data.items()):
        stdscr.addstr(height - SPARE_B, idx * 20 + 11, f"{value:<8}", curses.color_pair(2))
    stdscr.attroff(curses.color_pair(2))  # 값에 대한 색상 해제
    stdscr.refresh()

async def main(stdscr):
    curses.curs_set(0)  # 커서 숨기기
    stdscr.nodelay(True)  # getch() 호출이 블로킹되지 않도록 설정

    # 데이터 딕셔너리 예시
    data = {"CPU": "15%", "MEM": "34%", "DISK": "79%"}

    # 비동기 업데이트 작업 시작
    upper_terminal_task = asyncio.create_task(update_upper_terminal(stdscr))

    while True:
        # 하단 모니터링 바 그리기
        height, width = stdscr.getmaxyx()
        draw_monitoring_bar(stdscr, data, width, height)

        await asyncio.sleep(0.1)  # 이벤트 루프에 제어를 넘김

# curses.wrapper를 사용하여 asyncio 이벤트 루프와 함께 main 함수를 실행
curses.wrapper(lambda stdscr: asyncio.run(main(stdscr)))

 

 

출력결과 : 

 

---

코드 :

lib :  prompt_toolkit

key val 타입으로 입력받아서, 하단의 toolbar 형식으로 출력

밑에서 부터 출력시키는 형식

import asyncio
from prompt_toolkit import PromptSession
from prompt_toolkit.patch_stdout import patch_stdout


class AA():
  def __init__(self):
    self.k = 0
    self.v = 0
  def make_loop(self):
    self.main_loop = asyncio.get_event_loop()
    self.main_loop.run_until_complete(self.make_task())

  async def make_task(self):
    await asyncio.gather(
      self.ros_task_main(),
      self.another_task()
    )
    print('TASK END ERROR ', flush=True)

  async def another_task(self):
    while True:
      await asyncio.sleep(1)

  async def ros_task_main(self):
    asyncio.create_task(self.print_any())
    asyncio.create_task(self.checking_input())

  def update_toolbar_data(self, k, v):
    self.k = k
    self.v = v

  async def checking_input(self):
    def get_bottom_toolbar():
      return [
        ("", " "),
        ("bg:#000000 fg:#aaaa00", f"{self.k}"),
        ("", " : "),
        ("bg:#000000 fg:#00aaaa", f"{self.v}"),
        ("", ""),
      ]
    session = PromptSession()
    while True:
      with patch_stdout():
        session = PromptSession("", bottom_toolbar=get_bottom_toolbar)
        result = await session.prompt_async()
        # input_command = await session.prompt_async("", bottom_toolbar=get_bottom_toolbar)

  async def print_any(self):
    i = 0
    while True:
      print('count : {}'.format(i))
      i = i + 1
      await asyncio.sleep(0.1)


aa = AA()
aa.update_toolbar_data("emo", "y")
aa.make_loop()

 

 

출력결과 :