app_debug.py

from os import listdir
from os.path import isfile, join

import gradio as gr

import pipe
from io_utils import get_logs_file

LOG_PATH = "./tmp"
CONFIG_PATH = "./cicd/configs/"
MAX_FILES_NUM = 20


def get_accordions_of_files(path, files):
    components = [None for _ in range(0, MAX_FILES_NUM)]
    for i in range(0, len(files)):
        if i >= MAX_FILES_NUM:
            break
        with open(join(path, files[i]), "r") as f:
            components[i] = f.read()
    return components


def get_accordions_of_log_files():
    log_files = [
        f for f in listdir(LOG_PATH) if isfile(join(LOG_PATH, f)) and f.endswith("_log")
    ]
    return get_accordions_of_files(LOG_PATH, log_files)


def get_accordions_of_config_files():
    config_files = [
        f
        for f in listdir(CONFIG_PATH)
        if isfile(join(CONFIG_PATH, f)) and f.endswith(".yaml")
    ]
    return get_accordions_of_files(CONFIG_PATH, config_files)


def get_config_files():
    config_files = [
        join(CONFIG_PATH, f)
        for f in listdir(CONFIG_PATH)
        if isfile(join(CONFIG_PATH, f)) and f.endswith(".yaml")
    ]
    return config_files


def get_demo(demo):
    with gr.Row():
        # check if jobs is an attribute of pipe
        gr.Markdown(
            lambda: f"Current job: {pipe.current} Jobs in queue: {len(pipe.jobs)}"
            if hasattr(pipe, "jobs") and len(pipe.jobs) > 0
            else "No jobs in queue, please submit an evaluation task.",
            every=5,
        )
    with gr.Accordion(label="Log Files", open=False):
        logs = gr.Textbox(lines=10, visible=True, label="Log File")
        demo.load(get_logs_file, None, logs, every=0.5)
    with gr.Accordion(label="Config Files", open=False):
        gr.Files(value=get_config_files, label="Config Files", every=10)