jarvis-models/main.py

from typing import Annotated, Union

from fastapi import FastAPI, Request, status, Form
from fastapi.responses import JSONResponse

from src.dotchain.runtime.interpreter import program_parser
from src.dotchain.runtime.tokenizer import Tokenizer
from src.dotchain.runtime.runtime import Runtime
from src.blackbox.blackbox_factory import BlackboxFactory
import uvicorn
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


blackbox_factory = BlackboxFactory()

@app.post("/")
async def blackbox(blackbox_name: Union[str, None] = None, request: Request = None):
    if not blackbox_name:
        return await JSONResponse(content={"error": "blackbox_name is required"}, status_code=status.HTTP_400_BAD_REQUEST)
    try:
        box = blackbox_factory.create_blackbox(blackbox_name)
    except ValueError:
        return await JSONResponse(content={"error": "value error"}, status_code=status.HTTP_400_BAD_REQUEST)
    return await box.fast_api_handler(request)

def read_form_image(request: Request):
    async def inner(field: str):
        print(field)
        return "image"
    return inner

def read_form_text(request: Request):
    def inner(field: str):
        print(field)
        return "text"
    return inner

@app.post("/workflows")
async def workflows(script: Annotated[str, Form()], request: Request=None):
    dsl_runtime = Runtime(exteral_fun={"print": print, 
                                       'read_form_image': read_form_image(request),
                                       "read_form_text": read_form_text(request)})
    t = Tokenizer()
    t.init(script)
    ast = program_parser(t)
    ast.exec(dsl_runtime)

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, log_level="info")