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)


def print_interview(interview: prisma.models.InterviewStep):
    """
    Print the interview.
    """
    click.echo("-" * 40)
    click.echo(f"\033[92mSay:\033[0m {interview.say}")
    click.echo(f"\033[92mThoughts:\033[0m {interview.thoughts}")
    click.echo(
        f"\033[92mCreated At: {interview.createdAt.isoformat().split('.')[0]}\033[0m\n"
    )
    assert interview.Features, "Features is None"
    for feature in interview.Features:
        click.echo(f"\033[93m{feature.name}\033[0m")
        click.echo(f"\033[92mFunctionality:\033[0m {feature.functionality}")
        click.echo(f"\033[92mReasoning:\033[0m {feature.reasoning}\n")
    click.echo("-" * 40)


def print_spec_summary(spec: prisma.models.Specification):
    click.echo("-" * 40)
    click.echo("\033[92mSpecification Summary\033[0m")
    click.echo("\033[92mRequested Features:\33[0m")
    assert spec.Features, "Features is None"
    for feature in spec.Features:
        click.echo(f"- {feature.name}")

    assert spec.Modules, "Modules is None"
    route_cout = sum([len(module.ApiRouteSpecs) for module in spec.Modules])  # type: ignore
    module_count = len(spec.Modules)
    click.echo(
        f"\n\033[92m{module_count} Modules\33[0m with \033[93m{route_cout} routes\33[0m\n"
    )
    click.echo("-" * 40)

    for module in spec.Modules:
        click.echo(f"\033[92m- {module.name} -\033[0m {module.description}")
        assert module.ApiRouteSpecs, "ApiRouteSpecs is None"
        for route in module.ApiRouteSpecs:
            click.echo(
                f"  - \033[93m[{route.AccessLevel}]\033[0m \033[92m{route.method}\033[0m {route.path}"
            )
        click.echo("-" * 40)


def print_api_route_details(spec: prisma.models.Specification):
    click.echo("-" * 40)
    click.echo("\033[92mAPI Route Details\033[0m")
    assert spec.Modules, "Modules is None"

    for module in spec.Modules:
        click.echo(f"\033[93m- {module.name} -\033[0m {module.description}\n")

        assert module.ApiRouteSpecs, "ApiRouteSpecs is None"
        for route in module.ApiRouteSpecs:
            click.echo(f"\n\033[93m  - {route.method}\033[0m {route.path}")
            click.echo(f"\033[92m    Description:\033[0m {route.description}")
            click.echo(f"\033[92m    Access Level:\033[0m {route.AccessLevel}")
            click.echo(
                f"\033[92m    Allowed Access Roles:\033[0m {', '.join(route.AllowedAccessRoles).upper()}"
            )
            if route.RequestObject:
                click.echo(message="\033[92m    Request Object:\033[0m")
                if route.RequestObject.Fields is not None:
                    for param in route.RequestObject.Fields:
                        click.echo(
                            f"      - {param.name}: {param.typeName}  - {param.description}"
                        )
            if route.ResponseObject:
                click.echo("\033[92m    Response Object:\033[0m")
                if route.ResponseObject.Fields is not None:
                    for param in route.ResponseObject.Fields:
                        click.echo(
                            f"      - {param.name}: {param.typeName}  - {param.description}"
                        )
        click.echo("")
        click.echo("-" * 40)
        click.echo("")


def print_database_schema(spec: prisma.models.Specification):
    click.echo("-" * 40)
    click.echo("\033[92mDetailed Database Schema\033[0m")
    assert spec.DatabaseSchema, "DatabaseSchema is None"
    assert spec.DatabaseSchema.DatabaseTables, "DatabaseTables is None"
    for table in spec.DatabaseSchema.DatabaseTables:
        click.echo(f"\n{table.definition}\n")
    click.echo("-" * 40)
    click.echo("\033[92mDatabase Schema Summary\033[0m")
    for table in spec.DatabaseSchema.DatabaseTables:
        prefix = "\033[93m- Enum: " if table.isEnum else "\033[92m- Table:"
        click.echo(f"{prefix}\033[0m {table.name}")


async def explore_database(this: WhatToDebug):
    import codex.interview.database
    import codex.requirements.database

    prisma_client = prisma.Prisma()
    await prisma_client.connect()

    assert this.app_ids.applicationId, "Application ID is None"

    match this.object:
        case DebugObjects.APP:
            app = await prisma.models.Application.prisma().find_unique_or_raise(
                where={"id": this.app_ids.applicationId}
            )
            print_app(app, this.app_ids)
        case DebugObjects.INTERVIEW:
            assert this.app_ids.interviewId, "Interview ID is None"
            click.echo(f"What to debug: {this.app_ids}")
            interview_step = await codex.interview.database.get_last_interview_step(
                interview_id=this.app_ids.interviewId, app_id=this.app_ids.applicationId
            )
            print_interview(interview_step)
        case DebugObjects.SPEC_SUMMARY:
            assert this.app_ids.specificationId, "Specification ID is None"
            assert this.app_ids.userId, "User ID is None"
            spec = await codex.requirements.database.get_specification(
                this.app_ids.userId,
                app_id=this.app_ids.applicationId,
                spec_id=this.app_ids.specificationId,
            )
            print_spec_summary(spec)
        case DebugObjects.API_ROUTE_SPEC:
            assert this.app_ids.specificationId, "Specification ID is None"
            assert this.app_ids.userId, "User ID is None"
            spec = await codex.requirements.database.get_specification(
                this.app_ids.userId,
                app_id=this.app_ids.applicationId,
                spec_id=this.app_ids.specificationId,
            )
            print_api_route_details(spec)
        case DebugObjects.DATABASE_SCHEMA:
            assert this.app_ids.specificationId, "Specification ID is None"
            assert this.app_ids.userId, "User ID is None"
            spec = await codex.requirements.database.get_specification(
                this.app_ids.userId,
                app_id=this.app_ids.applicationId,
                spec_id=this.app_ids.specificationId,
            )
            print_database_schema(spec)
        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