app.py

from __future__ import annotations
from pathlib import Path
import gradio as gr
from typing import Any, Callable
import contextvars
from uw_programmatic.uw_machine import UWMachine
from uw_programmatic.csv_to_markdown import convert_csv_to_markdown
import pandas as pd
from dotenv import load_dotenv
import os

# Used to ensure that EventBus works across Gradio and the StateMachine.
def run_with_context(func: Callable) -> Callable:
    ctx = contextvars.copy_context()

    def wrapper(*args, **kwargs) -> Any:
        return ctx.run(func, *args, **kwargs)

    return wrapper

# Runs the state machine - sends the parameters over and kicks off the whole process.
def generate_questions(
    page_lower, page_higher, question_number, taxonomy
) -> tuple[str, gr.DownloadButton, gr.DownloadButton]:
    if machine.value and machine.value.current_state_value == "start":
        machine.value.start_machine()  # Start the machine!
    if not question_number or question_number <= 0:
        msg = "Choose a valid question number."
        raise gr.Error(msg)
    if not page_lower or not page_higher or page_higher < page_lower:
        msg = "Choose a valid page range."
        raise gr.Error(msg)
    if page_higher - page_lower <= 6:
        msg = "Page range must be >6."
        raise gr.Error(msg)
    if not taxonomy or len(taxonomy) == 0:
        msg = "Choose at least one taxonomy."
        raise gr.Error(msg)
    try:
        machine.value.send(
            "process_event",
            event_={
                "type": "user_input",
                "value": {
                    "page_range": (page_lower, page_higher),
                    "question_number": question_number,
                    "taxonomy": taxonomy,
                },
            },
        )
    except AttributeError:
        create_statemachine()
        machine.value.start_machine()
        machine.value.send(
            "process_event",
            event_={
                "type": "user_input",
                "value": {
                    "page_range": (page_lower, page_higher),
                    "question_number": question_number,
                    "taxonomy": taxonomy,
                },
            },
        )
    button = gr.DownloadButton(
        label="Download Questions",
        visible=True,
        scale=1,
        value=f"{Path.cwd().joinpath('outputs/professor_guide.csv')}",
    )
    button_2 = gr.DownloadButton(
        visible=True,
        value=f"{Path.cwd().joinpath('outputs/rejected_list.csv')}",
        scale=1,
        label="Download Rejected Questions",
    )
    markdown = "## Questions Ready for Download Below"
    if machine.value.errored:
        button = gr.update(visible=False)
        button_2 = gr.update(visible=False)
        markdown = "## Questions Not Ready for Download"
        exception = "Bad Page Range Selected"
        raise gr.Error(exception)
    # After questions are generated, read the CSV file
    csv_path = Path.cwd().joinpath("outputs/professor_guide.csv")
    if csv_path.exists():
        checkFrame = pd.read_csv(csv_path, usecols=[0], header=None)
        checkFrame.columns = ["Check"]
        if checkFrame.iloc[0]["Check"] != "Failed to generate more questions." and checkFrame.iloc[0]["Check"] != "Too many rejected questions.":
            # Generate markdown content
            markdown_content = convert_csv_to_markdown(csv_path)
            return (
                markdown,
                button,
                button_2,
                gr.update(value=markdown_content, visible=True)
            )
        else:
            return (markdown, button, button_2, gr.update(visible=False))
    else:
        return (markdown, button, button_2, gr.update(visible=False))


def create_statemachine() -> None:
    # Creates UWMachine from the config.yaml in current directory
    cwd_path = Path.cwd() / "uw_programmatic"
    config_path = cwd_path.joinpath(Path("config.yaml"))
    try:
        machine.value = UWMachine.from_config_file(config_path)
    except Exception as e:
        raise gr.Error(str(e)) from e


def questions_downloaded() -> dict[str, Any]:
    return gr.update(visible=False)


with gr.Blocks() as demo:
    gr.Markdown("# UW Quiz Generator")
    machine = gr.State(value=None)
    with gr.Row():
        with gr.Column(scale=2):
            taxonomy = gr.CheckboxGroup(
                choices=["Knowledge", "Comprehension", "Application"],
                label="Taxonomy",
                value="Knowledge",
            )
            question_number = gr.Number(
                minimum=1, maximum=25, label="Number of Questions", value=3
            )
            gr.Markdown("For Textbook - Pages 1-348")
            with gr.Row():
                page_lower = gr.Number(
                    label="First Page", minimum=1, value=1, maximum=348
                )
                page_higher = gr.Number(
                    label="Last Page", minimum=1, value=348, maximum=348
                )
            start_button = gr.Button(value="Generate Questions", scale=1)

        with gr.Column(scale=1):
            output = gr.Markdown("## Questions Not Ready for Download", visible=True)
            download_professor = gr.DownloadButton(
                label="Download Questions", visible=False, scale=1
            )
            download_failures = gr.DownloadButton(
                label="Download Failed Questions", visible=False, scale=1
            )
    with gr.Row():
        #Markdown component to display generated questions as an ordered list
        question_output = gr.Markdown(visible=False)
    # State machine must be initialized in gr.Blocks for the context wrapper to work.
    create_statemachine()
    start_button.click(
        fn=run_with_context(generate_questions),
        inputs=[page_lower, page_higher, question_number, taxonomy],
        outputs=[output, download_professor, download_failures, question_output],
    )
    download_professor.click(
        fn=run_with_context(questions_downloaded), outputs=[download_professor]
    )
    download_failures.click(
        fn=run_with_context(questions_downloaded), outputs=[download_failures]
    )

load_dotenv()
demo.launch(
    share=True,
    ssr_mode=False,
    auth=(os.environ.get("HF_USERNAME", ""), os.environ.get("HF_PASSWORD", "")),
)
#demo.launch()