babyagi/tools/monitor.py

#!/usr/bin/env python3
import sys
import time
import curses

from pathlib import Path
sys.path.append(str(Path(__file__).resolve().parent.parent))
from extensions.ray_objectives import CooperativeObjectivesListStorage
from extensions.ray_tasks import CooperativeTaskListStorage

def print_buffer(stdscr, lines):
    stdscr.clear()
    y = 0
    x = 0
    for line in lines:
        stdscr.addstr(y, x, line)
        y += 1
    stdscr.refresh()

def main(stdscr):
    objectives = CooperativeObjectivesListStorage()
    while True:
        objectives_list = objectives.get_objective_names()
        buffer = []
        if not objectives_list:
            buffer.append("No objectives")
        for objective in objectives_list:
            buffer.append("-----------------")
            buffer.append(f"Objective: {objective}")
            buffer.append("-----------------")
            tasks = CooperativeTaskListStorage(objective)
            tasks_list = tasks.get_task_names()
            buffer.append(f"Tasks:")
            for t in tasks_list:
                buffer.append(f" * {t}")
            buffer.append("-----------------")
        print_buffer(stdscr, buffer)
        time.sleep(30)

curses.wrapper(main)