Create app.py

app.py

@@ -0,0 +1,323 @@
+from operator import indexOf
+import os
+import gradio as gr
+import asyncio
+from dotenv import load_dotenv
+from pathlib import Path
+from time import sleep
+from glob import glob
+import copy
+import base64
+import json
+from PIL import Image
+from io import BytesIO
+import math
+import requests
+from urllib.parse import quote
+from multiprocessing import context
+from agents.tool import WebSearchTool
+from typing_extensions import TypedDict, Any
+from agents import Agent, FunctionTool, ImageGenerationTool, RunContextWrapper,  Runner, function_tool, SQLiteSession
+from markdown_pdf import MarkdownPdf
+from markdown_pdf import Section
+import brave
+import geo_distance
+
+load_dotenv(override=True)
+# key = os.getenv('OPENAI_API_KEY')
+users = os.getenv('LOGNAME')
+unames = users.split(',')
+pwds = os.getenv('PASSWORD')
+pwd_list = pwds.split(',')
+# DEEPSEEK_KEY=os.getenv('DEEPSEEK_KEY')
+# GROQ_KEY=os.getenv('GROQ_KEY')
+BRAVE_KEY=os.getenv('BRAVE_KEY')
+# BRAVE_SEARCH_KEY=os.getenv('BRAVE_SEARCH_KEY')
+LOCATIONID_KEY=os.getenv('LOCATIONID_KEY')
+
+# site = os.getenv('SITE')
+# if site == 'local':
+#     dp = Path('./data')
+#     dp.mkdir(exist_ok=True)
+#     dataDir = './data/'
+# else:
+#     dp = Path('/data')
+#     dp.mkdir(exist_ok=True)
+#     dataDir = '/data/'
+
+@function_tool
+async def make_pdf(text: str, title: str)->str:
+    '''Creates a pdf document based on input markdown text and title string.
+    
+    Args:
+        text: The markdown text to be processed to create the pdf document.
+        title:  A title that will be used as part of a filename in a file path.
+
+    Returns:
+        The file path to the PDF that was created and saved.
+    '''
+    for file in glob('./document.pdf'):
+        os.remove(file)
+    pdf = MarkdownPdf()
+    pdf.add_section(Section(text))
+    outpath = os.path.join('./','document.pdf')
+    pdf.save(outpath)
+    return outpath
+
+@function_tool
+async def get_news(query: str, window: str)->str:
+    '''Searches the internet news sources for news related to the query, within
+    a time window defined by argument window.
+
+    Args:
+        query: The topic about which news is desired.
+        window: a string indicating the time window over which news occurs, must be either 'day', 'week', 'month' or 'year'.
+    '''
+    periods = {"day":"pd","week":"pw","month":"pm","year":"py"}
+    window = window.casefold()
+    period = 'pw'
+    if window in periods.keys():
+        period = periods[window]
+    news = brave.get_brave_news(query, BRAVE_KEY, period)
+    return news
+
+@function_tool
+async def search_web(query: str)->str:
+    '''Searches the internet for information related to the query.
+
+    Args:
+        query: The topics to be searched on.
+    '''
+
+    result = brave.get_brave_search_results(query, BRAVE_KEY)
+    return result
+
+@function_tool
+async def get_distance(addr1: str, addr2: str)->float:
+    '''Compute the great-circle distance in miles between two addresses or other location specifiers.
+
+    Args:
+        addr1: The first address or location specifier.
+        addr2: The second address or location specifier.
+
+    Returns:
+        Distance in miles
+    '''
+
+    (lat1, lon1) = geo_distance.get_geo_coords(addr1, LOCATIONID_KEY)
+    (lat2, lon2) = geo_distance.get_geo_coords(addr2, LOCATIONID_KEY)
+    distance = geo_distance.great_circle_distance_miles(lat1, lon1, lat2, lon2)
+    return distance
+
+
+def md(txt):
+    return str(txt).replace('```', ' ').replace('  ', '&nbsp;&nbsp;').replace('  ', '&nbsp;&nbsp;').replace('  ', '&nbsp;&nbsp;').replace('\n','<br>').replace('~~','~')
+
+# def clean_history(db_path: str):
+#     connection = sqlite3.connect(db_path)
+#     cursor = connection.cursor()
+
+#     # Query to fetch all table names
+#     cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
+#     tables = cursor.fetchall()
+
+#     # Drop each table
+#     for table_name in tables:
+#         if table_name != 'sqlite_sequence':
+#             cursor.execute(f"DELETE FROM {table_name[0]};")
+
+#     # Commit changes and close the connection
+#     connection.commit()
+#     connection.close()
+
+def updatePassword(pwd, user):
+    password = 'none'
+    if user in unames:
+        pwd = pwd.lower().strip()
+        if pwd == pwd_list[unames.index(user)]:
+            password = pwd
+            return [password, "*********"]
+        else:
+            return [password, "invalid password"]
+    else:
+        return [password, "invalid user"]
+
+def update_user(user_win):
+    user_win = user_win.lower().strip()
+    user = 'unknown'
+    for s in unames:
+        if user_win == s:
+            user = s
+            break
+    return [user, user]
+
+def credentials_ok(user, pwd):
+    return user in unames and pwd in pwd_list
+
+def clean_up_files():
+    for file in glob('./document.pdf'):
+        try:
+            os.remove(file)
+        except:
+            pass
+    for file in glob('./*.png'):
+        try:
+            os.remove(file)
+        except:
+            pass
+
+def load_image(image, user):
+    status = 'OK, image is ready! Enter prompt and tap submit button'
+    try:
+        with open(image, 'rb') as image_file:
+            base64_image = base64.b64encode(image_file.read()).decode('utf-8')
+        fpath = user + '_image.b64'
+        with open(fpath, 'wt') as fp:
+            fp.write(base64_image)
+    except:
+        status = 'Unable to upload image'
+    return [fpath, status]
+
+def upload_image(user, password):
+    if not credentials_ok(user, password):
+        return [gr.Image(visible=False, interactive=True), "Incorrect user name and/or password"]
+    return [gr.Image(visible=True, interactive=True), '']
+
+async def chat(prompt_window, user_window, password, history, output_window,
+              uploaded_image_file):
+    file_download = gr.DownloadButton(label='Download File', visible=False, value=None)
+    image_window = gr.Image(visible=False, value=None)
+
+    if not credentials_ok(user_window, password):
+        return ['Invalid Credentials', prompt_window, uploaded_image_file,
+                     image_window, file_download, history]
+    instructions = '''
+    You are a helpful assistant.
+    You can call tools to compute straight-line distances and to search the web for
+    either information or news.  When you search for news you need to specify a period
+    of either 'day', 'week', 'month' or 'year'.  You also have a tool to create PDF 
+    documents and it requires markdown text as input.  If a distance is requested use
+    straight-line distance by default, and when possible use street addresses for locations.
+    '''
+
+    agent = Agent(name="Assistant",
+                 instructions=instructions,
+                 tools=[get_distance, search_web, get_news, make_pdf,
+              ImageGenerationTool(tool_config={"type": "image_generation", "quality": "low"},)],)
+
+    response = output_window
+    if not response:
+        response = ''
+    prompt = prompt_window
+    history.append({"role":"user", "content":prompt})
+    inputs = history
+    image_input = ''
+    if uploaded_image_file != '':
+        with open(uploaded_image_file, 'rt') as fp:
+            b64data = fp.read()
+        os.remove(uploaded_image_file)
+        uploaded_image_file = ''
+        image_input = (
+            {
+                "role": "user",
+                 "content": [
+                     {
+                         "type": "input_image",
+                         "image_url": f'data:image/jpeg;base64, {b64data}',
+                    }
+                ]
+            }
+        )
+        inputs.append(image_input)
+    result = await Runner.run(agent, max_turns=20,
+            input=inputs)
+    for item in result.new_items:
+        if (
+            item.type == "tool_call_item"
+            and item.raw_item.type == "image_generation_call"
+            and (img_result := item.raw_item.result)
+        ):
+            # with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp:
+            image_out_path = f'{user_window}_out.png'
+            with open(image_out_path,'wb') as fp:
+                fp.write(base64.b64decode(img_result))
+            image_window = gr.Image(visible=True, value=image_out_path)
+
+    history.append({"role":"assistant", "content":result.final_output})
+
+    reply = md(result.final_output)
+    response += "\n\n***YOU***: " + prompt + "\n\n***GPT***: " +  reply.replace('```','\n\n```\n\n')
+    usage = result.context_wrapper.usage
+    response += f"\nTotal tokens: = {usage.total_tokens}"
+    pdf_list = glob('./document.pdf')
+    if len(pdf_list):
+        file_download = gr.DownloadButton(label='Download PDF Doc',
+                                         visible=True, value='./document.pdf')
+    return [response, '', uploaded_image_file, image_window, file_download, history]
+
+# outputs=[history, output_window, prompt_window, uploaded_image_file,
+#                      image_window, file_download])
+
+def new_session(user_window, history):
+    history = []
+    return [prompt_window, history, 'Session cleared',
+           gr.Image(visible=False, value=None),
+           gr.Image(visible=False, value=None), '',
+           gr.DownloadButton(label='Download File', visible=False, value=None),
+           gr.File(visible=False, label='Upload File', type='filepath')]
+
+
+with gr.Blocks(theme=gr.themes.Soft()) as demo:
+    password = gr.State("")
+    user = gr.State("unknown")
+    uploaded_image_file = gr.State('')
+    uploaded_file_path = gr.State('')
+    history = gr.State([])
+
+    gr.Markdown('# GPT Agent')
+    gr.Markdown('Enter user name & password.  Tap "Help & Hints" button for more instructions.')
+    with gr.Row():
+        user_window = gr.Textbox(label = "User Name")
+        user_window.blur(fn=update_user, inputs=user_window, outputs=[user, user_window])
+        pwd_window = gr.Textbox(label = "Password")
+        help_button = gr.Button(value='Help & Hints')
+    with gr.Row():
+        clear_button = gr.Button(value="Start New Session")
+        button_do_image = gr.Button(value='Make Image')
+        button_upload_file = gr.Button(value='Upload Input File')
+        button_get_image = gr.Button(value='Upload Image to Analyze')
+        submit_button = gr.Button(value="Submit Prompt/Question")
+    with gr.Row():
+        prompt_window = gr.Textbox(label = "Prompt or Question", scale=7)
+    gr.Markdown('### **Dialog:**')
+    #output_window = gr.Text(container=True, label='Dialog')
+    output_window = gr.Markdown(container=True)
+    file_download = gr.DownloadButton(label='Download File', visible=False, value=None)
+    with gr.Row():
+        with gr.Column():
+            image_window2 = gr.Image(visible=False, interactive=True, label='Image to Analyze', type='filepath')
+        with gr.Column():
+            image_window = gr.Image(visible=False, label='Generated Image')
+    with gr.Row():
+        file_uploader = gr.File(visible=False, label='Upload File', type='filepath')
+    submit_button.click(chat,
+             inputs=[prompt_window, user_window, password, history, output_window,
+                    uploaded_image_file],
+             outputs=[output_window, prompt_window, uploaded_image_file,
+                     image_window, file_download, history])
+    clear_button.click(fn=new_session, inputs=[user_window, history],
+                      outputs=[prompt_window, history, output_window, image_window, image_window2,
+                      uploaded_image_file, file_download, file_uploader])
+    # image_window.change(fn=delete_image, inputs=[user])
+    # help_button.click(fn=show_help, outputs=output_window)
+    button_get_image.click(fn=upload_image,inputs = [user, password],
+                          outputs = [image_window2, output_window])
+    image_window2.upload(fn=load_image, inputs=[image_window2, user], outputs=[uploaded_image_file, output_window])
+    pwd_window.blur(updatePassword, inputs = [pwd_window, user], outputs = [password, pwd_window])
+    # button_upload_file.click(fn=upload_file, inputs=[user, password],
+    #                         outputs=[file_uploader, output_window])
+    # file_uploader.upload(fn=load_file, inputs=[file_uploader, user], outputs=[uploaded_file_path, output_window])
+# demo.launch(share=True, allowed_paths=[dataDir], ssr_mode=False)
+    # demo.load(delete_db_files)
+    demo.unload(clean_up_files)
+demo.launch(share=True, ssr_mode=False)