Minimal HTML GUI.

Session is stored in db
Mocked chatGpt calls, to test async loading.
Works local and docker.

eval_code.py

@@ -0,0 +1,77 @@
+import json
+import time
+
+"""
+We will handle here the code evaluation phase.
+
+"""
+import json
+
+def clean_prompt_answer(answer):
+    """
+    Chatgpt4 is ok, does not pollute the code but 3.5 encloses it in ```
+
+    :param answer:
+    :return:
+    """
+    cleaned = []
+    for l in answer.split("\n"):
+        if l.startswith("```"):
+            continue
+        else:
+            cleaned.append(l)
+    return "\n".join(cleaned)
+
+def parse_chatgpt_answer(ans_text):
+    try:
+        js_answer = json.loads(ans_text)
+    except:
+        # for now we dump the error in console
+        import traceback
+        exception = traceback.format_exc()
+        print(exception)
+        return {"error":exception}
+    return js_answer
+
+def eval_code_by_chatgpt(openai_client, ccode):
+    """
+    Will evaluate a piece of code using our heavily tuned prompt!
+
+    :param openai_client:
+    :param ccode:
+    :return:
+    """
+    time.sleep(3)
+    try:
+        return """[
+  {
+    "criteria": "DRY",
+    "explanation": "The memory allocation and initialization for ``p1``, ``p2``, and ``p3`` are repetitive. Consider creating a function like ``allocateAndInitializeMemory``."
+  },
+  {
+    "criteria": "SRP",
+    "explanation": "The ``main`` function handles memory allocation, initialization, and printing. You should separate these responsibilities into different functions like ``allocateMemory``, ``initializeData``, and ``printData``."
+  },
+  {
+    "criteria": "NAME",
+    "explanation": "``x1`` should be called ``title``, ``y1`` should be called ``author``, ``z1`` should be called ``year``, ``p1`` should be called ``titlePtr``, ``p2`` should be called ``authorPtr``, ``p3`` should be called ``yearPtr``."
+  }
+]"""
+        assert openai_client is not None
+    except:
+        import traceback
+        traceback.print_exc()
+        return {"error":"There was an error while parsing the answer. Maybe ChatGPT is overloaded?"}
+
+def eval_the_piece_of_c_code(openai_client, ccode):
+    """
+    Main entrypoint to this module. Will be called from backend
+
+    :param ccode:
+    :return:
+    """
+    chatgpt_ans = eval_code_by_chatgpt(openai_client, ccode)
+    chatgpt_js = parse_chatgpt_answer(chatgpt_ans)
+    return chatgpt_js
+
+

main.py

@@ -1,6 +1,13 @@
+import json
+import uuid
+
 from fasthtml.common import *
 import os
-import uuid
+import eval_code
+from datetime import datetime
+from sqlite_minutils.db import Database
+
+# the secrets. Will be loaded from HF, or for docker --env-file or from IDE
 
 OAUTH_CLIENT_ID = os.environ.get('OAUTH_CLIENT_ID')
 OAUTH_SCOPES = os.environ.get('OAUTH_SCOPES')
@@ -8,130 +15,267 @@ OAUTH_CLIENT_SECRET = os.environ.get('OAUTH_CLIENT_SECRET')
 OPENID_PROVIDER_URL = os.environ.get('OPENID_PROVIDER_URL')
 SPACE_HOST = os.environ.get('SPACE_HOST')
 
-app, rt = fast_app()
-
-js_hf_imports = \
-    """
-{
-    "imports": {
-        "@huggingface/hub": "https://cdn.jsdelivr.net/npm/@huggingface/hub@0.13.0/+esm"
-    }
-}
-"""
-
-js_block_hf_auth = \
-    """
-import { oauthLoginUrl, oauthHandleRedirectIfPresent } from "@huggingface/hub";
-console.log("huggingface env", window.huggingface);
-let oauthResult = localStorage.getItem("oauth");
-if (oauthResult) {
-  try {
-    oauthResult = JSON.parse(oauthResult);
-  } catch {
-    oauthResult = null;
-  }
-}
-
-oauthResult ||= await oauthHandleRedirectIfPresent();
-if (oauthResult) {
-  document.querySelector("pre").textContent = JSON.stringify(oauthResult, null, 2);
-  localStorage.setItem("oauth", JSON.stringify(oauthResult));
-  document.getElementById("signout").style.removeProperty("display");
-  document.getElementById("signout").onclick = async function() {
-    localStorage.removeItem("oauth");
-    window.location.href = window.location.href.replace(/\?.*$/, '');
-    window.location.reload();
-  }
-} else {
-  document.getElementById("signin").style.removeProperty("display");
-  document.getElementById("signin").onclick = async function() {
-    // prompt=consent to re-trigger the consent screen instead of silently redirecting
-    window.location.href = (await oauthLoginUrl({scopes: window.huggingface.variables.OAUTH_SCOPES})) + "&prompt=consent";
-  }
-}
-"""
-
-js_block_hf_auth2 = \
-    """
-import { oauthLoginUrl, oauthHandleRedirectIfPresent } from "@huggingface/hub";
-let oauthResult = localStorage.getItem("oauth");
-if (oauthResult) {
-  try {
-    oauthResult = JSON.parse(oauthResult);
-  } catch {
-    oauthResult = null;
-  }
-}
-
-console.log("OAuth result", oauthResult);
-oauthResult ||= await oauthHandleRedirectIfPresent();
-
-if (oauthResult) {
-  document.querySelector("pre").textContent = JSON.stringify(oauthResult, null, 2);
-  localStorage.setItem("oauth", JSON.stringify(oauthResult));
-  console.log("There is an oauth result", oauthResult);
-  document.getElementById("signout").style.removeProperty("display");
-  document.getElementById("signout").onclick = async function() {
-    localStorage.removeItem("oauth");
-    window.location.href = window.location.href.replace(/\?.*$/, '');
-    window.location.reload();
-  }
-    
-} else {
-    console.log("No OAuth result. Setting the loging button event");
-    document.getElementById("signin").style.removeProperty("display");
-    document.getElementById("signin").onclick = async function() {
-    // prompt=consent to re-trigger the consent screen instead of silently redirecting
-    window.location.href = window.hf_redirect_url;
-  }
-}
-"""
-
-
-def get_hf_user_data():
-    pass
-
-
-@rt('/', methods="get")
-def get():
-    global SPACE_HOST
-    if "localhost" in SPACE_HOST:
-        space_host = f"htpp://{SPACE_HOST}/"
-    else:
-        space_host = f"https://{SPACE_HOST}/"
-    hf_redirect_url_func = (
-        f"import {{ oauthLoginUrl, }} from '@huggingface/hub'; \n"
-        f"window.hf_redirect_url = await oauthLoginUrl({{clientId:'{OAUTH_CLIENT_ID}',redirectUrl:'{space_host}',"
-        f"scopes:'{OAUTH_SCOPES}'}}) + '&prompt=consent';\n" 
-        f"console.log(window.hf_redirect_url);")
-
-    header = Head(Title("C code reviewing"),
-                  Script(src="https://unpkg.com/es-module-shims@1.7.0/dist/es-module-shims.js"),
-                  Script(js_hf_imports, type="importmap"),
-                  )
-    content = Body(Div(P("Some content!")),
-                   A(f"Space host: {space_host}", href=space_host),
-                   Pre(""),
-                   Script(hf_redirect_url_func, type="module"),
-                   Img(src="https://huggingface.co/datasets/huggingface/badges/resolve/main/sign-in-with-huggingface-xl-dark.svg",
-                       alt="Sign in with Hugging Face",
-                       style="cursor: pointer; display: none;",
-                       _id="signin", name=None),
-                   Button("Sign out", _id="signout", style="display: none", name=None),
-                   A("Authenticate!", href="authorized"),
-                   Script(js_block_hf_auth2, type="module"))
-    full_page = Html(header, content)
-    return full_page
-
-
-@rt('/change')
-def get(): return P('Nice to be here!')
-
-
-@rt('/authorized', methods="get")
-def authorized():
-    content = Body(P("Do we have authenticated user?"))
-    return content
+# The database. In memory bc it is efemerial on HF anyway.
+
+global_database = Database(memory=True, memory_name="mem_db_name")
+global_database_tables = global_database.t
+session_state_table = global_database_tables.session_state
+if session_state_table not in global_database_tables:
+    # session_id stored in cookies
+    # see EVAL_STATUS_x below for state
+    session_state_table.create(id=int, session_id=str, state=int, submitted=datetime, completed=datetime, answer=str, pk='id')
+Session_State_cls = session_state_table.dataclass()
+
+EVAL_STATE_NEW=0
+EVAL_STATE_QUERY=1
+EVAL_STATE_TIMEDOUT=2
+EVAL_STATE_ANSWER=3
+EVAL_STATE_ERROR=4
+
+
+if "localhost" in SPACE_HOST:
+    # Are we hacking locally?
+    print("Localhost detected in SPACE_HOST. App started in debug+live mode!")
+    app, rt = fast_app(debug=True, live=True)
+else:
+    app, rt = fast_app(debug=False, live=False)
+
+
+CODE_AUGMENTATIONS=[
+    ("DRY", "Don't repeat yourself."),
+    ("SRP", "Single object[function] responsability"),
+    ("MC", "Magic constants."),
+    ("NAME", "Meaningful names in the code."),
+]
+
+def format_code_review(code_js, html_id=""):
+    list_of_citerias = []
+    for caug_code, caug_txt in CODE_AUGMENTATIONS:
+        crit_tag = [H3(caug_code), P(caug_txt)]
+        list_of_citerias.extend(crit_tag)
+        # yeah, I know . . .
+        criteria_found = False
+        for eval_line in code_js:
+            if caug_code == eval_line["criteria"]:
+                eval_txt = Pre(eval_line["explanation"])
+                list_of_citerias.append(eval_txt)
+                criteria_found = True
+        if criteria_found is False:
+            list_of_citerias.append(P("Not infringed"))
+    return Div(*list_of_citerias, _id=html_id)
+
+def generate_default_eval_section(html_id=""):
+    return Div(P("This is where criterias will show up once the code is evaluated"), _id=html_id)
+
+def generate_working_eval_section(html_id=""):
+    return Div(P("Working  . . ."), _id=html_id,
+    hx_get=f"/render_answer",
+    hx_trigger = "every 2s",
+    hx_swap = "outerHTML",
+    )
+
+# How can I timeout? Well ... TBD.
+@threaded
+def call_gpt_and_store_result(pkid, code_to_check):
+    # print("evaluatign code")
+    try:
+        # Pesky way to get a new cursor, in a thread safe way, into the db. This code runs in another thread.
+        # Can we do better?
+        local_database = Database(memory=True, memory_name="mem_db_name")
+        local_cursor = local_database.t.session_state
+        Session_State_cls = local_cursor.dataclass()
+        local_sess_obj_lst = local_cursor(limit=1, where=f"id == {pkid}")
+        local_sess_obj = local_sess_obj_lst[0]
+        gpt_js_eval = eval_code.eval_the_piece_of_c_code(openai_client=None, ccode=code_to_check)
+        # print("done eval with ", gpt_js_eval)
+        # new_sess_obj = Session_State_cls(session_obj)
+        if "error" in gpt_js_eval:
+            local_sess_obj.state = EVAL_STATE_ERROR
+            local_sess_obj.answer = gpt_js_eval["error"]
+            local_sess_obj.completed = datetime.utcnow()
+        else:
+            local_sess_obj.state = EVAL_STATE_ANSWER
+            local_sess_obj.answer = json.dumps(gpt_js_eval)
+            local_sess_obj.completed = datetime.utcnow()
+        local_cursor.update(local_sess_obj)
+    except:
+        import traceback
+        traceback.print_exc()
+
+def get_rendered_answer(session_id):
+    state_rows = session_state_table(limit=1, where=f"session_id == '{session_id}'", order_by="id DESC")
+    # print(state_rows)
+    if len(state_rows) <= 0:
+        return generate_default_eval_section(html_id="prompt_response")
+    state = state_rows[0]
+    if state.state == EVAL_STATE_ANSWER:
+        gpt_js_eval = json.loads(state.answer)
+        return format_code_review(gpt_js_eval, html_id="prompt_response")
+    if state.state == EVAL_STATE_QUERY:
+        return generate_working_eval_section(html_id="prompt_response")
+    return Div(P("There was an error:", P(state.answer)), _id="prompt_response")
+
+
+@rt("/render_answer")
+def get(session):
+    if 'session_id' not in session: return "No session ID"
+    session_id = session["session_id"]
+    return get_rendered_answer(session_id)
+
+@rt("/")
+def get(session):
+    if 'session_id' not in session:
+        session['session_id'] = str(uuid.uuid4())
+    inp = Textarea(id="ccodetoeval", name="ccodetoeval", placeholder="Enter a piece of C code", rows=20)
+    # eval_area = Pre("", id="prompt_response")
+    # eval_area = format_code_review([], html_id="prompt_response")
+    eval_area = generate_default_eval_section(html_id="prompt_response")
+    add = Form(Group(inp, Button("Generate")), hx_post="/submit_to_eval", hx_swap="outerHTML",  target_id='prompt_response')
+    return Title('Image Generation Demo'), Main(H1('Magic Image Generation'), add, eval_area, cls='container')
+
+
+@rt("/submit_to_eval", methods="post")
+def post(ccodetoeval:str, session):
+    if 'session_id' not in session: return P("Bad call. No session ID")
+    session_id = session["session_id"]
+    session_obj = Session_State_cls(
+        session_id=session_id,
+        state=EVAL_STATE_QUERY,
+        submitted=datetime.utcnow(),
+        answer=""
+    )
+    # we insert and we get the new primary key
+    session_obj = session_state_table.insert(session_obj)
+    # will be executed in another thread with magic @threaded
+    call_gpt_and_store_result(session_obj.id, ccodetoeval)
+    return get_rendered_answer(session_id)
+
+
+## This code is for reference, for OAuth. BIG PITA, will be added later.
+
+#
+# js_hf_imports = \
+#     """
+# {
+#     "imports": {
+#         "@huggingface/hub": "https://cdn.jsdelivr.net/npm/@huggingface/hub@0.13.0/+esm"
+#     }
+# }
+# """
+#
+# js_block_hf_auth = \
+#     """
+# import { oauthLoginUrl, oauthHandleRedirectIfPresent } from "@huggingface/hub";
+# console.log("huggingface env", window.huggingface);
+# let oauthResult = localStorage.getItem("oauth");
+# if (oauthResult) {
+#   try {
+#     oauthResult = JSON.parse(oauthResult);
+#   } catch {
+#     oauthResult = null;
+#   }
+# }
+#
+# oauthResult ||= await oauthHandleRedirectIfPresent();
+# if (oauthResult) {
+#   document.querySelector("pre").textContent = JSON.stringify(oauthResult, null, 2);
+#   localStorage.setItem("oauth", JSON.stringify(oauthResult));
+#   document.getElementById("signout").style.removeProperty("display");
+#   document.getElementById("signout").onclick = async function() {
+#     localStorage.removeItem("oauth");
+#     window.location.href = window.location.href.replace(/\?.*$/, '');
+#     window.location.reload();
+#   }
+# } else {
+#   document.getElementById("signin").style.removeProperty("display");
+#   document.getElementById("signin").onclick = async function() {
+#     // prompt=consent to re-trigger the consent screen instead of silently redirecting
+#     window.location.href = (await oauthLoginUrl({scopes: window.huggingface.variables.OAUTH_SCOPES})) + "&prompt=consent";
+#   }
+# }
+# """
+#
+# js_block_hf_auth2 = \
+#     """
+# import { oauthLoginUrl, oauthHandleRedirectIfPresent } from "@huggingface/hub";
+# let oauthResult = localStorage.getItem("oauth");
+# if (oauthResult) {
+#   try {
+#     oauthResult = JSON.parse(oauthResult);
+#   } catch {
+#     oauthResult = null;
+#   }
+# }
+#
+# console.log("OAuth result", oauthResult);
+# oauthResult ||= await oauthHandleRedirectIfPresent();
+#
+# if (oauthResult) {
+#   document.querySelector("pre").textContent = JSON.stringify(oauthResult, null, 2);
+#   localStorage.setItem("oauth", JSON.stringify(oauthResult));
+#   console.log("There is an oauth result", oauthResult);
+#   document.getElementById("signout").style.removeProperty("display");
+#   document.getElementById("signout").onclick = async function() {
+#     localStorage.removeItem("oauth");
+#     window.location.href = window.location.href.replace(/\?.*$/, '');
+#     window.location.reload();
+#   }
+#
+# } else {
+#     console.log("No OAuth result. Setting the loging button event");
+#     document.getElementById("signin").style.removeProperty("display");
+#     document.getElementById("signin").onclick = async function() {
+#     // prompt=consent to re-trigger the consent screen instead of silently redirecting
+#     window.location.href = window.hf_redirect_url;
+#   }
+# }
+# """
+#
+#
+# def get_hf_user_data():
+#     pass
+
+
+# @rt('/', methods="get")
+# def get():
+#     global SPACE_HOST
+#     if "localhost" in SPACE_HOST:
+#         space_host = f"htpp://{SPACE_HOST}/"
+#     else:
+#         space_host = f"https://{SPACE_HOST}/"
+#     hf_redirect_url_func = (
+#         f"import {{ oauthLoginUrl, }} from '@huggingface/hub'; \n"
+#         f"window.hf_redirect_url = await oauthLoginUrl({{clientId:'{OAUTH_CLIENT_ID}',redirectUrl:'{space_host}',"
+#         f"scopes:'{OAUTH_SCOPES}'}}) + '&prompt=consent';\n"
+#         f"console.log(window.hf_redirect_url);")
+#
+#     header = Head(Title("C code reviewing"),
+#                   Script(src="https://unpkg.com/es-module-shims@1.7.0/dist/es-module-shims.js"),
+#                   Script(js_hf_imports, type="importmap"),
+#                   )
+#     content = Body(Div(P("Some content!")),
+#                    A(f"Space host: {space_host}", href=space_host),
+#                    Pre(""),
+#                    Script(hf_redirect_url_func, type="module"),
+#                    Img(src="https://huggingface.co/datasets/huggingface/badges/resolve/main/sign-in-with-huggingface-xl-dark.svg",
+#                        alt="Sign in with Hugging Face",
+#                        style="cursor: pointer; display: none;",
+#                        _id="signin", name=None),
+#                    Button("Sign out", _id="signout", style="display: none", name=None),
+#                    A("Authenticate!", href="authorized"),
+#                    Script(js_block_hf_auth2, type="module"))
+#     full_page = Html(header, content)
+#     return full_page
+
+
+# @rt('/change')
+# def get(): return P('Nice to be here!')
+
+
+# @rt('/authorized', methods="get")
+# def authorized():
+#     content = Body(P("Do we have authenticated user?"))
+#     return content
 
 
 serve()

requirements.txt

@@ -1 +1,2 @@
 python-fasthtml
+fastsql==1.0.1

test/test_dbops.py

@@ -0,0 +1,27 @@
+from fasthtml.common import *
+from sqlite_minutils.db import Database
+def test_some_sqldb_inmem():
+
+    db = Database(memory=True)
+    assert db is not None
+    db_tables = db.t
+    session_states = db_tables.session_states
+    if session_states not in db.t:
+        session_states.create(prompt=str, session_id=str, id=int, folder=str, pk='id')
+    Session_state = session_states.dataclass()
+    sess = Session_state(prompt="Some prompt", folder="Somewhere")
+    session_states.insert(sess)
+    session_states.insert(sess)
+    session_states.insert(sess)
+    sess2 = Session_state(prompt="Some prompt", folder="Somewhere new")
+    # insert AND get the newly inserted ID
+    g = session_states.insert(sess2)
+    assert g.id > 3
+    g.prompt = "New prompt"
+    # Update and check if it can be retrieved
+    session_states.update(g)
+    # retrieval
+    newg = session_states(where=f"id={g.id}")
+    assert len(newg) == 1
+    assert newg[0].prompt == "New prompt"
+

test/test_eval_code.py

@@ -0,0 +1,28 @@
+import pytest
+import eval_code
+
+TYPICAL_ANSWER = """[
+  {
+    "criteria": "DRY",
+    "explanation": "The memory allocation and initialization for ``p1``, ``p2``, and ``p3`` are repetitive. Consider creating a function like ``allocateAndInitializeMemory``."
+  },
+  {
+    "criteria": "SRP",
+    "explanation": "The ``main`` function handles memory allocation, initialization, and printing. You should separate these responsibilities into different functions like ``allocateMemory``, ``initializeData``, and ``printData``."
+  },
+  {
+    "criteria": "NAME",
+    "explanation": "``x1`` should be called ``title``, ``y1`` should be called ``author``, ``z1`` should be called ``year``, ``p1`` should be called ``titlePtr``, ``p2`` should be called ``authorPtr``, ``p3`` should be called ``yearPtr``."
+  }
+]"""
+
+
+def test_parse_chatgpt_answer():
+    ans = eval_code.parse_chatgpt_answer(TYPICAL_ANSWER)
+    assert ans is not None
+    assert len(ans) > 0
+    an_eval = ans[0]
+    assert "criteria" in an_eval.keys()
+    assert "explanation" in an_eval.keys()
+
+