call pgai and logger

app.py

@@ -5,17 +5,15 @@ from zoneinfo import ZoneInfo
 
 import gradio as gr
 from extract import extract
-import app_util
-from pgsoft.pgconst.const import service_list, functionality_list
-from pgsoft.pghost import ais
+import call_logger
 from pgsoft.pgdate.date_utils import beijing
 import call_pgai
-from str_util import normalize_text
+from utils import normalize_text
 
 #######################
 # proxy version
 #######################
-proxy_version = "1.0.0-2024-05-14-a"
+proxy_version = "1.0.0-2024-07-30-a"
 
 t = datetime.now()
 t = t.astimezone(ZoneInfo("Asia/Shanghai"))
@@ -26,84 +24,50 @@ print(f"[Seattle]: {t.replace(microsecond=0)}")
 
 identity = os.environ.get("identity")
 print(f"identity: {identity}")
-if not identity:
-    identity = "watermelon"
-ai = "stevez-ai-dev"
-if identity in ais:
-    ai = ais[identity]
 db_token = os.environ.get("db_token")
 if db_token:
     print(db_token[:5])
 
+game_list = [
+    "matchn",
+    "house",
+    "watermelon",
+    "snake",
+]
 
-def run(hf_token, service, game, functionality, nlp_command):
-    """
-    event handler
-    """
 
+def run(info, game, nlp_command):
+    """event handler"""
     # reuse hf_token field as json string
-    token, user, redirect, source, username, _ = extract(hf_token)
-    if user is None:
-        user = "__fake__"
-
-    # redirect all traffic to the proxy sever
-    global ai
-    if redirect is not None:
-        ai = redirect
-    ai_url = f"https://{ai}.hf.space"
-    if token is None or token == "":
-        return "please specify hf token"
-
-    if service not in service_list[1:]:
-        if game is None:
-            return "please specify which game"
-        if functionality is None:
-            return "please choose the AI functionality"
-        if functionality == "AI":
-            if nlp_command in ["", None]:
-                return "please make sure the command is not empty"
+    user, source, username, _ = extract(info)
+    if nlp_command is None:
+        return "command is required"
+    nlp_command = normalize_text(nlp_command)
+    if nlp_command == "":
+        return "invalid command"
 
     service_start = beijing()
-    print(f"<<<<<<<<<<<<<< service starts at {service_start} <<<<<<<<<<<<<<")
-    if service in ["download game", "upload game", "list games"]:
-        res = app_util.file_service(service, nlp_command, db_token)
-        if res is None:
-            outp = {"status": "Failure"}
-        else:
-            outp = {"status": "OK", "result": res}
-    else:
-        assert "games" in service_list
-        if service == "games":
-            print(f"{beijing()} [{user}] [{game}] {nlp_command}")
-        nlp_command = normalize_text(nlp_command)
-        call_pgai.from_cache = True
-        outp = call_pgai.call_pgai(
-            service,
-            game,
-            functionality,
-            nlp_command,
-            ai_url,
-            token,
-        )
-        if outp is None:
-            return "no output"
-        if isinstance(outp, str):
-            return outp
-        # add proxy version info to the output
-        outp["timestamp"] = beijing().__str__()
-        outp["proxy-version"] = proxy_version
-        outp["user"] = user
-        outp["username"] = username
-        outp["game"] = game
-        outp["source"] = source
-        outp["cache"] = call_pgai.from_cache
-        app_util.call_logger(outp, identity, token)
+    print(f"[{service_start}] service starts >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")
+    print(f"[{beijing()}] [{user}] [{game}] [{nlp_command}]")
+
+    call_pgai.from_cache = True
+    outp = call_pgai.call_pgai(nlp_command, game)
+    if outp is None:
+        return "no output"
+    if isinstance(outp, str):
+        return outp
+    # add proxy version info to the output
+    outp["timestamp"] = beijing().__str__()
+    outp["proxy-version"] = proxy_version
+    outp["user"] = user
+    outp["username"] = username
+    outp["game"] = game
+    outp["source"] = source
+    outp["cache"] = call_pgai.from_cache
+    call_logger.call_logger(outp, identity, db_token)
     service_end = beijing()
     timecost = service_end.timestamp() - service_start.timestamp()
-    print(
-        f">>>>>>>>>>>>>>> service ends at {service_end}, "
-        + f"costs {timecost:.2f}s >>>>>>>>>>>>>>>\n"
-    )
+    print(f"[{service_end}] service ends, costs {timecost:.2f}s <<<<<<<<<<<<<<<<<<<<<")
     return json.dumps(outp, indent=4)
 
 
@@ -112,25 +76,14 @@ demo = gr.Interface(
     inputs=[
         "text",
         gr.Radio(
-            service_list,
-            value=service_list[0],
-            info="Shared services",
-        ),
-        gr.Radio(
-            ["house"],
-            value="house",
+            game_list,
+            value=game_list[0],
             info="Which game you want the AI to support?",
         ),
-        gr.Radio(
-            functionality_list[:1],
-            value=functionality_list[0],
-            # label = "What do you want to do?",
-            info="What functionality?",
-        ),
         "text",
     ],
     outputs="text",
-    title="Demo",
+    title="Pgai",
     allow_flagging="never",
 )
 

app_util.py

@@ -1,154 +0,0 @@
-import json
-
-from gradio_client import Client
-from pgsoft.pgdate.date_utils import beijing
-from pgsoft.pgfile import download, upload, list_files
-from pgsoft.pghash.md5 import md5
-from time import sleep
-from huggingface_hub import HfApi
-import os
-
-
-def call_logger(log_info, caller, hf_token) -> None:
-    #######################
-    # logging
-    #######################
-    calling_start = beijing()
-    print(f"calling logger starts at {beijing()}")
-    #################################################
-    urls = [
-        "https://hubei-hunan-logger.hf.space",
-        "https://hubei-hunan-logger2.hf.space",
-    ]
-    for url in urls:
-        try:
-            client = Client(
-                url,
-                hf_token=hf_token,
-                verbose=False,
-            )
-            client.submit(json.dumps(log_info), caller)
-            print(f"[logging to {url}] OK")
-        except Exception as e:
-            print(f"[logging to {url}] error: {e}")
-    #################################################
-    calling_end = beijing()
-    timecost = calling_end.timestamp() - calling_start.timestamp()
-    print(f"calling logger ends at {calling_end}, costs {timecost:.2f}s")
-
-
-dataset_id = "pgsoft/game"
-tempdir = "game"
-gamename = "house"
-localdir = os.path.join(tempdir, gamename)
-if not os.path.exists(localdir):
-    os.makedirs(localdir)
-hf_api = HfApi()
-
-
-def file_service(service, arg: str, token: str):
-    """download game, upload game, or list games"""
-    if service == "download game":
-        filename = arg.strip() + ".json"
-        remotepath = "/".join(["house", filename[:2], filename])
-        res = download(
-            dataset_id,
-            remotepath=remotepath,
-            repo_type="dataset",
-            localdir=tempdir,
-            token=token,
-        )
-        if not res:
-            return None
-        with open(res, "r") as f:
-            outp = json.load(f)
-        print(f"[{service}] OK")
-        return outp
-    elif service == "upload game":
-        try:
-            game = json.loads(arg)
-        except json.JSONDecodeError as e:
-            print(f"[{service}] {type(e)}: {e}")
-            return None
-
-        if not isinstance(game, dict):
-            print(f"[{service}] not a dict")
-            return None
-
-        needed_keys = ["game-file", "device-id"]
-        for key in needed_keys:
-            if key not in game:
-                print(f'[{service}] error: missed "{key}"')
-                return None
-        if not isinstance(game["device-id"], str):
-            print(f'[{service}] error: "device-id" is not a str')
-            return None
-        if not isinstance(game["game-file"], dict):
-            print(f'[{service}] error: "game-file" is not a dict')
-            return None
-
-        obj = {
-            "upload-time": beijing().__str__(),
-            "game-file": game["game-file"],
-        }
-
-        maxtry = 5
-        for retry in range(maxtry):
-            md5code = md5(obj)
-            remotepath = "/".join([gamename, md5code[:2], md5code + ".json"])
-            if not hf_api.file_exists(
-                repo_id=dataset_id,
-                filename=remotepath,
-                repo_type="dataset",
-                token=token,
-            ):
-                break
-            sleep(0.1)
-            obj["upload-time"] = beijing().__str__()
-            maxtry -= 1
-        if not maxtry and hf_api.file_exists(
-            repo_id=dataset_id,
-            filename=remotepath,
-            repo_type="dataset",
-            token=token,
-        ):
-            print(f"[{service}] error: file exists")
-            return None
-        filedir = os.path.join(localdir, md5code[:2])
-        if not os.path.exists(filedir):
-            os.mkdir(filedir)
-        filepath = os.path.join(filedir, md5code + ".json")
-        content = json.dumps(game, indent=4)
-        with open(filepath, "w") as f:
-            f.write(content)
-        res = upload(
-            filepath,
-            remotepath,
-            dataset_id,
-            "dataset",
-            token,
-            f"Updated at {beijing()}",
-        )
-        if not res:
-            print(f"[{service}] error: upload failed")
-            return None
-        print(f"[{service}] OK")
-        return md5code
-    elif service == "list games":
-        games = list_files(
-            repo_id=dataset_id,
-            repo_type="dataset",
-            token=token,
-        )
-        if games is None:
-            return None
-        games = {
-            item.split(".")[0][-32:]: item
-            for item in games
-            if item.endswith(".json") and item.startswith("house")
-        }
-        print(f"[{service}] OK")
-        return games
-    else:
-        print(f"[{service}] error: unknown service")
-        return None

call_ai.py

@@ -1,37 +0,0 @@
-from functools import cache
-from gradio_client import Client
-from pgsoft.pgdate.date_utils import beijing
-import json
-
-from_cache = True
-
-
-@cache
-def call_ai(service, game, functionality, nlp_command, url, hf_token):
-    calling_start = beijing()
-    print(f"calling ai starts at {calling_start}")
-    try:
-        client = Client(
-            url,
-            hf_token=hf_token,
-            verbose=False,
-        )
-        res = client.predict(
-            service,
-            game,
-            functionality,
-            nlp_command,  # hidden,
-            api_name="/predict",
-        )
-    except Exception as e:
-        return (
-            f"{type(e)}, {str(e)}. \nyou may want to make "
-            + "sure your hf_token is correct"
-        )
-    calling_end = beijing()
-    timecost = calling_end.timestamp() - calling_start.timestamp()
-    print(f"calling ai ends at {calling_end}, costs {timecost:.2f}s")
-    outp = json.loads(res)
-    global from_cache
-    from_cache = False
-    return outp

call_logger.py

@@ -0,0 +1,32 @@
+import json
+
+from gradio_client import Client
+from pgsoft.pgdate.date_utils import beijing
+
+
+def call_logger(log_info, caller, hf_token) -> None:
+    #######################
+    # logging
+    #######################
+    calling_start = beijing()
+    print(f"[{calling_start}] logging starts")
+    #################################################
+    urls = [
+        "https://hubei-hunan-logger.hf.space",
+        "https://hubei-hunan-logger2.hf.space",
+    ]
+    for url in urls:
+        try:
+            client = Client(
+                url,
+                hf_token=hf_token,
+                verbose=False,
+            )
+            client.submit(json.dumps(log_info), caller)
+            print(f"[logging to {url}] OK")
+        except Exception as e:
+            print(f"[logging to {url}] error: {e}")
+    #################################################
+    calling_end = beijing()
+    timecost = calling_end.timestamp() - calling_start.timestamp()
+    print(f"[{calling_end}] logging ends, costs {timecost:.2f}s")

call_pgai.py

@@ -6,14 +6,14 @@ from pgsoft.pgdate.date_utils import beijing
 import json
 
 
-def post_to_pgai_helper(command: str) -> Any:
+def post_to_pgai_helper(command: str, game: str) -> Any:
     myobj = {
-        "code": os.getenv("pgai_code"), 
+        "code": os.getenv("pgai_code"),
         "command": command,
     }
 
-    header = {'accept': 'application/json', 'Content-Type': 'application/json'}
-    url_base = "https://steveagi-pgai.hf.space/games/house/nlp"
+    header = {"accept": "application/json", "Content-Type": "application/json"}
+    url_base = f"https://steveagi-pgai.hf.space/games/{game}/nlp"
     url_read = f"{url_base}/r"
 
     try:
@@ -27,27 +27,27 @@ def post_to_pgai_helper(command: str) -> Any:
     except Exception as e:
         print(e)
         return None
-    
+
 
 from_cache = True
+
+
 # @cache
-def call_pgai(service, game, functionality, nlp_command, url, hf_token):
+def call_pgai(nlp_command, game):
     calling_start = beijing()
-    print(f"calling ai starts at {calling_start}")
+    print(f"[{calling_start}] calling ai starts")
     try:
-        res = post_to_pgai_helper(nlp_command)
+        res = post_to_pgai_helper(nlp_command, game)
     except Exception as e:
-        return (
-            f"{type(e)}, {str(e)}. \nyou may want to make "
-            + "sure your hf_token is correct"
-        )
+        return f"{type(e)}: {str(e)}."
+
     calling_end = beijing()
     timecost = calling_end.timestamp() - calling_start.timestamp()
-    print(f"calling ai ends at {calling_end}, costs {timecost:.2f}s")
+    print(f"[{calling_end}] calling ai ends, costs {timecost:.2f}s")
     global from_cache
     from_cache = False
-    if res is not None:
-        outp = json.loads(res)
-        return outp
-    else:
-        return None
+    if res is None:
+        return res
+    outp = json.loads(res)
+    return outp
+

extract.py

@@ -1,20 +1,14 @@
 import json
 
 
-def extract(hf_token):
-    """
-    Extract token, user, redirect, source, username and info from input hf_token.
-    If hf_token is simple, it is the token itself.
-    """
+def extract(info_str):
+    """Extract user, source, username, and all info"""
     info = {}  # a copy of hf_token in json format
     try:
-        info = dict(json.loads(hf_token))
+        info = dict(json.loads(info_str))
     except json.decoder.JSONDecodeError:
-        return hf_token, None, None, None, None,None
-    token = info.get("token", None)
-    user = info.get("user", None)
-    redirect = info.get("redirect", None)
+        return None, None, None, None
+    user = info.get("user", "__fake__")
     source = info.get("source", None)
     username = info.get("username", None)
-
-    return token, user, redirect, source, username, info
+    return user, source, username, info