0.0.1.2 included common app files and utilities.

application.py

@@ -0,0 +1,76 @@
+import os
+
+'''
+shared environment variables
+'''
+default_domain = "cervical-cage"
+
+aws_access_key_id = os.environ.get('AMRA_AWS_ACCESS_KEY_ID')
+aws_secret_access_key = os.environ.get('AMRA_AWS_SECRET_ACCESS_KEY')
+openai_api_key = os.environ.get('AMRA_OPENAI_API_KEY')
+
+'''
+ui device environment variables
+'''
+eu_device_classification = ["phone", "laptop", "tablet", "desktop"]
+performance_outcome_list = [ "VAS Score", "Incidence of Fusion", "Incidence of Pseudoarthrosis", "Incidence of Reoperation"]
+safety_outcome_list = ["Incidence of Revision","Incidence of Nonunion"]
+
+device_options={
+    "secondary extraction":False,
+    "secondary extraction count":0
+}
+
+'''
+ui equivalent environment variables
+'''
+ec_options={
+    "Equivalent Comparator":False,
+    "Equivalent Comparator require SD":False,
+    "Equivalent Comparator count":0
+}
+
+'''
+dynamodb tables structure
+'''
+data_structure = {
+    "terms":{
+        "key":{
+
+        },
+        "fields":[
+        "assessment_step",
+        "term",
+        "clinical study",
+        "summary",
+        "command"
+    ]},
+    "prompts":[
+        "assessment_step",
+        "template_name",
+        "instruction"
+    ],
+    "articles":[
+        "domain",
+        "name",
+        "s3_path",
+        "vector",
+        "meta",
+        "content"
+    ],
+    "outputs":[
+        "domain",
+        "article",
+        "output"
+    ]
+}
+
+'''
+application default data
+'''
+app_data = {
+    "articles":[],
+    "terms":[],
+    "prompts":[],
+    "outputs":[]
+}

cloud_db.py

@@ -0,0 +1,69 @@
+import boto3
+
+from utility import *
+
+# initialize dynamodb instance
+db_client = boto3.client(
+    "dynamodb",
+    aws_access_key_id = aws_access_key_id,
+    aws_secret_access_key = aws_secret_access_key
+    )
+
+'''
+dynamodb data operations
+'''
+
+# get the list of articles from articles table in dynamodb
+def get_table(table_name:str):
+    result = db_client.scan(TableName = table_name,AttributesToGet = data_structure[table_name])
+    return db_list_to_py_list(result["Items"])
+
+# add a new article to table articles in dynamodb, return error if failed
+def post_item(table_name:str,item:dict):
+    try:
+        res = db_client.put_item(
+            TableName = table_name,
+            Item = py_dict_to_db_map(item)
+        )
+    except Exception as e:
+        return {"Error":e}
+    return res
+
+# update an article in table articles in dynamodb, return error if failed
+def put_item(table_name:str,item:dict):
+    try:
+        res = db_client.put_item(
+            TableName = table_name,
+            Item = py_dict_to_db_map(item)
+        )
+    except Exception as e:
+        return {"Error":e}
+    return res
+
+# delete an article in table articles in dynamodb, return error if not found.
+def delete_item(table_name:str,key:dict):
+    try:
+        res = db_client.delete_item(
+            TableName = table_name,
+            Key = py_dict_to_db_map(key)
+        )
+    except Exception as e:
+        return {"Error":e}
+    return res
+
+def get_item(table_name:str,key:dict):
+    try:
+        res = db_client.get_item(
+            TableName = table_name,
+            Key = py_dict_to_db_map(key)
+        )
+    except Exception as e:
+        return {"Error":e}
+    return res
+
+'''
+dynamodb structure management
+'''
+def get_structure(table_name:str):
+    result = db_client.describe_table(TableName = table_name)
+    return result["Table"]["AttributeDefinitions"]

invoices.py

@@ -0,0 +1,8 @@
+from cloud_db import *
+
+def get_invoices(user_id:str):
+    return db_map_to_pd_dataframe(get_table("invoices",{"user_id":user_id}))
+
+def get_invoice(user_id:str,invoice_id:str):
+    return get_item("invoices",{"user_id":user_id,"invoice_id":invoice_id})
+

user.py

@@ -0,0 +1,10 @@
+from cloud_db import *
+
+def get_user(user_id:str):
+    return get_item("users",{"user_id":user_id})
+
+def put_billing_info(user_id:str,billing_info:dict=None,payment_method:dict=None):
+    if not billing_info and not payment_method:
+        return {"Error":"No billing info or payment method provided"}
+        
+    return put_item("users",{"user_id":user_id,"billing_info":billing_info,"payment_method":payment_method})

utility.py

@@ -0,0 +1,163 @@
+import json
+import os
+# import PyPDF2
+import pandas as pd
+
+from application import *
+
+'''
+following functions are for file manipulation
+'''
+
+# read pdf file and return text
+def read_pdf(file_path):
+    # open the pdf file
+    try:
+        filename = file_path
+        pdfFileObj = open(file_path, 'rb')    
+    except TypeError:
+        filename = file_path.name
+        pdfFileObj = open(file_path.name, 'rb')
+
+    # create a pdf reader object
+    pdfReader = PyPDF2.PdfReader(pdfFileObj)
+
+    # get the number of pages in the pdf file
+    num_pages = len(pdfReader.pages)
+
+    # create an empty string
+    text = ''
+
+    # iterate through all the pages
+    for page_num in range(num_pages):
+        page_obj = pdfReader.pages[page_num]
+        text += page_obj.extract_text ()
+
+    # close the pdf file object
+    pdfFileObj.close()
+
+    text = remove_symbols(text)
+
+    with open(f"{filename.split('.')[0]}.txt", "w") as f:
+        f.write(text)
+
+    # return the string of text
+    return text, pdfReader.metadata
+
+'''
+following functions are for format standard response
+'''
+
+# format standard response for status code and data
+def format_response(code,data):
+    return {
+        "statusCode":code,
+        "headers":{
+            "Access-Control-Allow-Origin": "*",
+            "Content-Type": "application/json"
+        },
+        "body":json.dumps(data),
+        "isBase64Encoded": False
+    }
+
+'''
+following functions are for string manipulation
+'''
+
+# format text output by removing excessive characters
+def format_text(text,remove_char_ls = ["\\n--\\n","\\n\\n","\n"]):
+    for c in remove_char_ls:
+        text = text.replace(c,"")
+    
+    return text
+
+# function to remove symbols that are not in unicode
+def remove_symbols(text):
+    # remove symbols that are not in unicode
+    text = text.encode("ascii", "ignore").decode()
+    # remove the break word new line return
+    text = text.replace('-\n', '')
+    return text
+
+def str_to_tuple(s):
+    return tuple(s.replace("(","").replace(")","").split(","))
+
+'''
+following functions are for dynamodb data manipulation
+'''
+# convert dynamodb map to python dictionary
+def db_map_to_py_dict(db_map):
+    py_dict = {}
+    for k,i in db_map.items():
+        for l,v in i.items():
+            if l == "M":
+                py_dict[k] = db_map_to_py_dict(v)
+            elif l == "S": 
+                py_dict[k] = v
+            elif l == "N":
+                py_dict[k] = int(v) if float(v)%1 ==0 else float(v)
+            elif l == "L":
+                py_dict[k] = db_list_to_py_list(v)
+            else:
+                py_dict[k] = v
+        
+    return py_dict
+
+def db_map_to_pd_dataframe(db_map):
+    py_dict = db_map_to_py_dict(db_map)
+    return pd.DataFrame(py_dict)
+
+# convert python dictionary to dynamodb map
+def py_dict_to_db_map(py_dict):
+    db_map = {}
+    for key,value in py_dict.items():
+        key = str(key)
+        if type(value) is str:
+            db_map[key] = {"S":value}
+        elif type(value) is int or type(value) is float:
+            db_map[key] = {"N":value}
+        elif type(value) is dict:
+            db_map[key] = {"M":py_dict_to_db_map(value)}
+        elif type(value) is list:
+            db_map[key] = {"L":py_list_to_db_list(value)}
+            
+    return db_map
+
+# convert dynamodb list to python list
+def db_list_to_py_list(db_list):
+    py_list = []
+    for d in db_list:
+        for t,v in d.items():
+            if t == "M":
+                py_list.append(db_map_to_py_dict(v))
+            elif t == "L":
+                py_list.append(db_list_to_py_list(v))
+            else:
+                py_list.append(v)
+                
+    return py_list
+
+# convert python list to dynamodb list
+def py_list_to_db_list(py_list):
+    db_list = []
+    for value in py_list:
+        if type(value) is str:
+            item = {"S":value}
+        elif type(value) is int or float:
+            item = {"N":value}
+        elif type(value) is dict:
+            item = {"M":py_dict_to_db_map(value)}
+        elif type(value) is list:
+            item = {"L":py_list_to_db_list(value)}
+        
+        db_list.append(item)
+        
+    return db_list
+
+'''
+following functions are used for business logic. (to be moved to business logic layer)
+'''
+
+# function to calculate the estimated cost of the translation
+def est_cost(n_tokens,rate):
+    return round(rate*n_tokens/1000,4)