AutoGPT-Code-Ability/codex/debug/__init__.py

import asyncio
import enum

import click
import prisma
import pydantic

import codex.common.model
import codex.debug.queries


@click.group()
def debug():
    """
    Tools for debugging the codex system.
    """
    pass


class DebugObjects(enum.Enum):
    """
    Possible objects to debug.
    """

    APP = "App"
    INTERVIEW = "Interview"
    SPEC_SUMMARY = "Spec Summary"
    API_ROUTE_SPEC = "API Route Spec"
    DATABASE_SCHEMA = "Database Schema"
    FUNCTION = "Function"
    COMPILED_ROUTE = "Compiled Route"
    COMPLETED_APP = "Completed App"
    DEPLOYMENT = "Deployment"


class WhatToDebug(pydantic.BaseModel):
    """
    What to debug.
    """

    object: DebugObjects
    app_ids: codex.common.model.ResumePoint


async def get_resume_points(prisma_client) -> list[codex.common.model.ResumePoint]:
    import datetime

    if not prisma_client.is_connected():
        await prisma_client.connect()

    resume_points_list = await codex.debug.queries.fetch_application_details()
    resume_points = resume_points_list.resume_points

    print(
        "\033[92m Todays\033[0m and\033[93m yesterdays\033[0m days apps are colored:\n"
    )
    print(
        f"{'':<3} | {'updatedAt':<20} | {'name':<30} | {'Interview':<10} | {'Specification':<15} | {'CompiledAApp':<12} | {'Deployment':<10}"
    )
    separation_row = f"{'-' * 3}-+-{'-' * 20}-+-{'-' * 30}-+-{'-' * 10}-+-{'-' * 15}-+-{'-' * 12}-+-{'-' * 10}"
    print(separation_row)
    # Print table rows
    today = datetime.date.today()
    yesterday = today - datetime.timedelta(days=1)

    for i, resume_point in enumerate(resume_points):
        updated_at_date = datetime.datetime.fromisoformat(
            resume_point.updatedAt.isoformat().split(".")[0]
        ).date()

        # Determine the color based on the updated date
        if updated_at_date == today:
            color_code = "\033[92m"  # Green
        elif updated_at_date == yesterday:
            color_code = "\033[93m"  # Yellow
        else:
            color_code = "\033[0m"  # Reset to default
        row = [
            str(i + 1),
            resume_point.updatedAt.isoformat().split(".")[0],
            resume_point.name[:30],
            "✓" if resume_point.interviewId else "X",
            "✓" if resume_point.specificationId else "X",
            "✓" if resume_point.completedAppId else "X",
            "✓" if resume_point.deploymentId else "X",
        ]
        formatted_row = f"{color_code}{row[0]:<3} | {row[1]:<20} | {row[2]:<30} | {row[3]:<10} | {row[4]:<15} | {row[5]:<12} | {row[6]:<10}\033[0m"
        print(formatted_row)
    print(
        "\n"
        + " " * 59
        + f"{resume_points_list.pagination.current_page}/{resume_points_list.pagination.total_pages}"
    )

    return resume_points


def what_to_debug():
    """
    Works out what the user would like to debug.
    """
    click.echo("What app would you like to debug?")
    click.echo("Here are the latest apps and where they got up to.")

    prisma_client = prisma.Prisma(auto_register=True)
    click.echo("")
    resume_points = asyncio.get_event_loop().run_until_complete(
        get_resume_points(prisma_client)
    )
    click.echo("\n")
    case = int(input("Select index of the app you want to debug: "))
    app = resume_points[case - 1]
    assert app is not None, "App is None"
    click.echo("Great, now what are you intestested in debugging?")
    click.echo("Here are the different objects you can inspect:\n")
    for i, object in enumerate(DebugObjects):
        click.echo(f"{i + 1}. {object.value}")

    case = int(input("\nSelect index of the object you want to debug: "))
    click.echo(
        f"Okay, lets see what is going on with {app.name}'s {list(DebugObjects)[case -1].value}"
    )

    return WhatToDebug(object=list(DebugObjects)[case - 1], app_ids=app)


def print_app(
    app: prisma.models.Application, resume_point: codex.common.model.ResumePoint
):
    """
    Print the app.
    """
    click.echo(f"\033[92m\nName: {app.name}\033[0m")
    click.echo(f"\033[92mDescription: {app.description}\033[0m")
    click.echo(
        f"\033[92mCreated At: {app.createdAt.isoformat().split('.')[0]}\033[0m\n"
    )
    click.echo("-" * 40)
    click.echo("✓ Interview" if resume_point.interviewId else "X Interview")
    click.echo("✓ Specification" if resume_point.specificationId else "X Specification")
    click.echo("✓ Compiled App" if resume_point.completedAppId else "X Compiled App")
    click.echo("✓ Deployment" if resume_point.deploymentId else "X Deployment")
    click.echo("-" * 40)


async def explore_database(this: WhatToDebug):
    prisma_client = prisma.Prisma()
    await prisma_client.connect()
    match this.object:
        case DebugObjects.APP:
            app = await prisma.models.Application.prisma().find_unique(
                where={"id": this.app_ids.applicationId}
            )
            if app:
                print_app(app, this.app_ids)
        case DebugObjects.INTERVIEW:
            pass
        case DebugObjects.SPEC_SUMMARY:
            pass
        case DebugObjects.API_ROUTE_SPEC:
            pass
        case DebugObjects.DATABASE_SCHEMA:
            pass
        case DebugObjects.FUNCTION:
            pass
        case DebugObjects.COMPILED_ROUTE:
            pass
        case DebugObjects.COMPLETED_APP:
            pass
        case DebugObjects.DEPLOYMENT:
            pass
        case _:
            pass
    await prisma_client.disconnect()


@debug.command()
@click.option("--port", "-p", default=8000, help="The port to debug.")
def llmcalls(port: int):
    """
    Debug the LLM calls.
    """
    pass


@debug.command()
@click.option("--port", "-p", default=8000, help="The port to debug.")
def object(port: int):
    """
    Debug the database objects.
    """
    this = what_to_debug()
    asyncio.get_event_loop().run_until_complete(explore_database(this))

    pass