""" title: Make charts out of your data author: Omar EL HACHIMI author_url: https://github.com/OM-EL?tab=repositories author_linkedin: https://www.linkedin.com/in/omar-el-hachimi-b48286158/ funding_url: https://github.com/open-webui revisingauthor_url: https://github.com/CutterSol version: 0.0.2r3 Note1: untested update to Omar's orginal code. Given the information in the document titled Migration Guide: Open WebUI 0.4 to 0.5. Note2: If you have trouble running this function, open a python cli and try doing the imports in the first few lines. If the imports error out then use pip to install the modules. """ from pydantic import BaseModel, Field from typing import Optional, Dict, Any import os from open_webui.models.files import Files import uuid import logging from openai import OpenAI import time # Set up logging logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger(__name__) SYSTEM_PROMPT_BUILD_CHARTS = """ Objective: Your goal is to read the query, extract the data, choose the appropriate chart to present the data, and produce the HTML to display it. Steps: 1. Read and Examine the Query: • Understand the user’s question and identify the data provided. 2. Analyze the Data: • Examine the data in the query to determine the appropriate chart type (e.g., bar chart, pie chart, line chart) for effective visualization. 3. Generate HTML: • Create the HTML code to present the data using the selected chart format. 4. Handle No Data Situations: • If there is no data in the query or the data cannot be presented as a chart, generate a humorous or funny HTML response indicating that the data cannot be presented. 5. Calibrate the chart scale based on the data: • based on the data try to make the scale of the chart as readable as possible. Key Considerations: - Your output should only include HTML code, without any additional text. - Generate only HTML. Do not include any additional words or explanations. - Make to remove any character other non alpha numeric from the data. - is the generated HTML Calibrate the chart scale based on the data for eveything to be readable. - Generate only html code , nothing else , only html. Example1 : ''' Interactive Chart

''' Example2: ''' Collaborateurs par Métier/Fonction

''' 2. No Data or Unchartable Data: '''

We''re sorry, but your data can''t be charted.

Maybe try feeding it some coffee first?

''' """ USER_PROMPT_GENERATE_HTML = """ Giving this query {Query} generate the necessary html qurty. """ # Initialize OpenAI client class FileData(BaseModel): id: str filename: str meta: Dict[str, Any] class Action: class Valves(BaseModel): show_status: bool = Field( default=True, description="Show status of the action." ) html_filename: str = Field( default="json_visualizer.html", description="Name of the HTML file to be created or retrieved.", ) OPENIA_KEY: str = Field( default="", description="key to consume OpenIA interface like LLM for example a litellm key.", ) OPENIA_URL: str = Field( default="", description="Host where to consume the OpenIA interface like llm", ) def __init__(self): self.valves = self.Valves() self.openai = None self.html_content = """ """ def create_or_get_file(self, user_id: str, json_data: str) -> str: filename = str(int(time.time() * 1000)) + self.valves.html_filename directory = "action_embed" logger.debug(f"Attempting to create or get file: {filename}") # Check if the file already exists existing_files = Files.get_files() for file in existing_files: if ( file.filename == f"{directory}/{user_id}/{filename}" and file.user_id == user_id ): logger.debug(f"Existing file found. Updating content.") # Update the existing file with new JSON data self.update_html_content(file.meta["path"], json_data) return file.id # If the file doesn''t exist, create it base_path = os.path.join("uploads", directory) os.makedirs(base_path, exist_ok=True) file_path = os.path.join(base_path, filename) logger.debug(f"Creating new file at: {file_path}") self.update_html_content(file_path, json_data) file_id = str(uuid.uuid4()) meta = { "source": file_path, "title": "Modern JSON Visualizer", "content_type": "text/html", "size": os.path.getsize(file_path), "path": file_path, } # Create a new file entry file_data = FileData( id=file_id, filename=f"{directory}/{user_id}/{filename}", meta=meta ) new_file = Files.insert_new_file(user_id, file_data) logger.debug(f"New file created with ID: {new_file.id}") return new_file.id def update_html_content(self, file_path: str, html_content: str): with open(file_path, "w", encoding="utf-8") as f: f.write(html_content) logger.debug(f"HTML content updated at: {file_path}") async def action( self, body: dict, __user__=None, __event_emitter__=None, __event_call__=None, ) -> Optional[dict]: logger.debug(f"action:{__name__} started") await __event_emitter__( { "type": "status", "data": { "description": "Analysing Data", "done": False, }, } ) if __event_emitter__: try: original_content = body["messages"][-1]["content"] self.openai = OpenAI( api_key=self.valves.OPENIA_KEY, base_url=self.valves.OPENIA_URL ) response = self.openai.chat.completions.create( model="bedrock/anthropic.claude-3-sonnet-20240229-v1:0", messages=[ {"role": "system", "content": SYSTEM_PROMPT_BUILD_CHARTS}, { "role": "user", "content": USER_PROMPT_GENERATE_HTML.format( Query=body["messages"][-1]["content"] ), }, ], max_tokens=1000, n=1, stop=None, temperature=0.7, ) html_content = response.choices[0].message.content print("-----------------------------") # print html content in pretty and readable format # this is to help debug print(html_content) print("-----------------------------") user_id = __user__["id"] file_id = self.create_or_get_file(user_id, html_content) # Create the HTML embed tag html_embed_tag = f"{{{{HTML_FILE_ID_{file_id}}}}}" # Append the HTML embed tag to the original content on a new line body["messages"][-1][ "content" ] = f"{original_content}\n\n{html_embed_tag}" await __event_emitter__( { "type": "status", "data": { "description": "Visualise the chart", "done": True, }, } ) logger.debug(f" objects visualized") except Exception as e: error_message = f"Error visualizing JSON: {str(e)}" logger.error(f"Error: {error_message}") body["messages"][-1]["content"] += f"\n\nError: {error_message}" if self.valves.show_status: await __event_emitter__( { "type": "status", "data": { "description": "Error Visualizing JSON", "done": True, }, } ) logger.debug(f"action:{__name__} completed") return body