import enum import logging import ssl import urllib from itertools import groupby from typing import List, Union, Optional from urllib.error import URLError from dearpygui import dearpygui as dpg from graph_exploring_tool import query from graph_exploring_tool.configuration import Configuration from graph_exploring_tool.graphical.constants import STATUS_CONSOLE_TAG, QUERY_FILTER_TAG, EDITOR_SELECTOR_TAG, \ ENDPOINT_TEXTINPUT_TAG, QUERY_EDITOR_VISUAL_TAG, POST_METHOD_CHECKBOX_TAG, QUERY_EDITOR_TEXTUAL_TAG, \ QUERY_RESULT_WINDOW_TAG, QUERY_RESULT_TAB_TAG, RESULT_TABLE_TAG, RADIO_BUTTON_TYPE, MAIN_QUERY_GROUP_TAG from graph_exploring_tool.query import QueryTemplate class StatusMessageType(enum.Enum): DEFAULT = 0 WARNING = 1 ERROR = 2 def _config(sender, keyword, user_data): widget_type = dpg.get_item_type(sender) items = user_data if widget_type == RADIO_BUTTON_TYPE: value = True else: keyword = dpg.get_item_label(sender) value = dpg.get_value(sender) if isinstance(user_data, list): for item in items: dpg.configure_item(item, **{keyword: value}) else: dpg.configure_item(items, **{keyword: value}) def set_status_text(message: str, message_type: StatusMessageType = StatusMessageType.DEFAULT): """Set the text displayed on the status bar.""" dpg.set_value(STATUS_CONSOLE_TAG, message) with dpg.theme() as default_theme: with dpg.theme_component(dpg.mvAll): dpg.add_theme_color(dpg.mvThemeCol_Text, (255, 255, 255), category=dpg.mvThemeCat_Core) with dpg.theme() as warning_theme: with dpg.theme_component(dpg.mvAll): dpg.add_theme_color(dpg.mvThemeCol_Text, (50, 200, 50), category=dpg.mvThemeCat_Core) with dpg.theme() as error_theme: with dpg.theme_component(dpg.mvAll): dpg.add_theme_color(dpg.mvThemeCol_Text, (200, 50, 50), category=dpg.mvThemeCat_Core) if message_type == StatusMessageType.DEFAULT: dpg.bind_item_theme(STATUS_CONSOLE_TAG, default_theme) elif message_type == StatusMessageType.WARNING: dpg.bind_item_theme(STATUS_CONSOLE_TAG, warning_theme) elif message_type == StatusMessageType.ERROR: dpg.bind_item_theme(STATUS_CONSOLE_TAG, error_theme) def create_query_palette(query_palette: List[QueryTemplate]): with dpg.child_window(label="Query Palette", width=220, menubar=True): with dpg.menu_bar(): dpg.add_menu(label="Query Palette", enabled=False) dpg.add_input_text(label="Filter", callback=lambda s, a: dpg.set_value(QUERY_FILTER_TAG, a)) with dpg.filter_set(tag=QUERY_FILTER_TAG): # TODO Populate from onto/base # TODO Fix filtering grouped_query_palette = groupby(sorted(query_palette, key=lambda palette_query: palette_query.group), key=lambda palette_query: palette_query.group) for query_template_group, query_template_queries in grouped_query_palette: with dpg.tree_node(label=query_template_group, tag=query_template_group, default_open=True, filter_key=""): full_filter = "" for query_template in query_template_queries: query_button = dpg.add_button(label=query_template.name, filter_key=query_template.name, callback=_query_palette_click, user_data=query_template) with dpg.tooltip(dpg.last_item()): dpg.add_text(query_template.description) full_filter =f"{full_filter},{query_template.name}" # TODO Make theme-ing more generic with dpg.theme() as item_theme: with dpg.theme_component(dpg.mvAll): dpg.add_theme_color(dpg.mvThemeCol_Text, (200, 200, 100), category=dpg.mvThemeCat_Core) dpg.add_theme_style(dpg.mvStyleVar_FrameRounding, 0, category=dpg.mvThemeCat_Core) with dpg.theme() as modifies_theme: with dpg.theme_component(dpg.mvAll): dpg.add_theme_color(dpg.mvThemeCol_Text, (200, 50, 50), category=dpg.mvThemeCat_Core) if query_template.visual_support: dpg.bind_item_theme(query_button, item_theme) if query_template.modifies: dpg.bind_item_theme(query_button, modifies_theme) dpg.configure_item(query_template_group, filter_key=full_filter) def create_query_options(endpoint: str): with dpg.group(horizontal=True): dpg.add_combo(("Visual", "Textual"), default_value="Visual", callback=_mode_select, width=100, tag=EDITOR_SELECTOR_TAG) dpg.add_input_text(default_value=endpoint, width=500, enabled=True, tag=ENDPOINT_TEXTINPUT_TAG) dpg.add_checkbox(label="debug", callback=_config) dpg.add_checkbox(label="annotate", default_value=False, callback=_config) # TODO Actually make this annotate the result data with its type # TODO Make the elements color coded etc... # TODO Fix context menu and keyboard operations def create_query_editor_visual(example_prefix:str, example_query:str, show: bool = False): with dpg.child_window(autosize_x=True, height=500, menubar=True, show=show, tag=QUERY_EDITOR_VISUAL_TAG): with dpg.menu_bar(): dpg.add_menu(label="Visual Query Editor", enabled=False) with dpg.tab_bar(): with dpg.tab(label="Query", tag="__query_tab"): dpg.add_input_text(default_value=example_query, multiline=True, on_enter=True, height=300, width=-1, tag="__query_editor_visual_input") with dpg.tab(label="Prefix", tag="__prefix_tab"): dpg.add_input_text(default_value=example_prefix, multiline=True, on_enter=True, height=300, width=-1, tag="__prefix_editor_visual_input") with dpg.tab(label="Info", tag="__info_tab"): # TODO Add information about the query. Like what it does etc... dpg.add_text("", tag="__query_info_label") with dpg.tab(label="Advanced", tag="__advanced_tab"): dpg.add_checkbox(label="Use post method", default_value=False, tag=POST_METHOD_CHECKBOX_TAG) with dpg.group(tag="__visual_editor_fields"): # TODO Make more clear which placeholder is getting replaced (Maybe do it interactively, replace as user changes) pass with dpg.group(horizontal=True): dpg.add_button(label="Query", callback=_perform_query) dpg.add_button(label="Save", callback=_select_directory) # TODO dpg.add_button(label="Load") # TODO # TODO Maybe remove/change this option, we made all the queries mostly 'visual' # It's practically a duplicate at this point def create_query_editor_textual(example_prefix:str, example_query:str, show: bool = False): with dpg.child_window(autosize_x=True, height=500, menubar=True, show=show, tag=QUERY_EDITOR_TEXTUAL_TAG): with dpg.menu_bar(): dpg.add_menu(label="Textual Query Editor", enabled=False) with dpg.tab_bar(): with dpg.tab(label="Query", tag="__query_textual_tab"): dpg.add_input_text(default_value=example_query, multiline=True, on_enter=True, height=300, width=-1, tag="__query_editor_textual_input") with dpg.tab(label="Prefix", tag="__prefix_textual_tab"): dpg.add_input_text(default_value=example_prefix, multiline=True, on_enter=True, height=300, width=-1, tag="__prefix_editor_textual_input") with dpg.group(horizontal=True): dpg.add_button(label="Query", callback=_perform_query) dpg.add_button(label="Save", callback=_select_directory) # TODO dpg.add_button(label="Load") # TODO # TODO Not at the bottom because fok getting that layout to fit correctly def create_status_console(): with dpg.child_window(autosize_x=True, height=35): with dpg.group(horizontal=True): dpg.add_text(default_value="Ready.", tag=STATUS_CONSOLE_TAG) def set_copy(element: Union[int, str], text_data:str): dpg.set_value(element, shorten(text_data)) i = dpg.get_item_label(element) dpg.set_value(f"__copy_{i}", shorten(text_data)) dpg.configure_item(f"__copy_drag_payload_{i}", drag_data=shorten(text_data)) dpg.set_value(f"__copy_drag_{i}", shorten(text_data)) def create_query_results(): with dpg.child_window(autosize_x=True, autosize_y=True, menubar=True, tag=QUERY_RESULT_WINDOW_TAG): with dpg.menu_bar(): dpg.add_menu(label="Results", enabled=False) # dpg.add_input_text(label="Filter", tag="__result_filter_input", callback=lambda s, a: dpg.set_value("__result_filter", a)) with dpg.tab_bar(): with dpg.tab(label="Query Result", tag=QUERY_RESULT_TAB_TAG): with dpg.table(header_row=True, policy=dpg.mvTable_SizingFixedFit, row_background=True, reorderable=True, resizable=True, no_host_extendX=False, hideable=True, borders_innerV=True, delay_search=True, borders_outerV=True, borders_innerH=True, borders_outerH=True, tag=RESULT_TABLE_TAG): pass with dpg.tab(label="Saved"): with dpg.group(horizontal=True): with dpg.group(): for i in range(0, 10): dpg.add_input_text(label=f"{i}", payload_type="string", width=500, drop_callback=lambda s, a: set_copy(s,a)) with dpg.group(): for i in range(0, 10): dpg.add_text(tag=f"__copy_{i}", default_value="Empty") with dpg.drag_payload(parent=dpg.last_item(), drag_data="Empty", payload_type="string", tag=f"__copy_drag_payload_{i}"): dpg.add_text(tag=f"__copy_drag_{i}", default_value="Empty") with dpg.tab(label="Debug"): dpg.add_text("This is the debug tab!") def _mode_select(sender: int, mode: str, user_data: Optional[dict]): mode = mode.lower().strip() visual = mode == "visual" # TODO When showing visual we need to fix the custom fields dpg.configure_item(QUERY_EDITOR_VISUAL_TAG, show=visual) dpg.configure_item(QUERY_EDITOR_TEXTUAL_TAG, show=not visual) def shorten(identifier: str) -> str: for key, value in query.reverse_prefix.items(): identifier = identifier.replace(key, f"{value}:") return identifier def _query_palette_click(sender: int, mode: str, user_data: Optional[QueryTemplate]): mode = dpg.get_value(EDITOR_SELECTOR_TAG).lower().strip() if mode == "visual" and not user_data.visual_support: logging.warning(f"Visual mode for template '{user_data.name}' not implemented yet!") set_status_text(f"Visual mode for template '{user_data.name}' not implemented yet!", StatusMessageType.WARNING) return set_status_text(f"Using {mode} template '{user_data.name}'.") dpg.set_value(f"__prefix_editor_{mode}_input", user_data.prefix) dpg.set_value(f"__query_editor_{mode}_input", user_data.query) dpg.set_value(POST_METHOD_CHECKBOX_TAG, user_data.modifies) dpg.delete_item("__visual_editor_fields", children_only=True) if mode == "visual" and user_data.visual_support: for replacement in user_data.replacements: dpg.add_input_text(label=f"{replacement.description}",default_value=replacement.suggestion, payload_type="string", width=500, parent="__visual_editor_fields", drop_callback=lambda s, a: dpg.set_value(s, shorten(a)), tag=f"__{replacement.suggestion}", user_data=replacement) def _perform_query(sender: int, mode: str, user_data: Optional[dict]): # TODO Fix for new query structure mode = dpg.get_value(EDITOR_SELECTOR_TAG).lower().strip() prefix_text = dpg.get_value(f"__prefix_editor_{mode}_input") query_text = dpg.get_value(f"__query_editor_{mode}_input") endpoint = dpg.get_value(ENDPOINT_TEXTINPUT_TAG) use_post_method = dpg.get_value(POST_METHOD_CHECKBOX_TAG) try: if mode=="visual": for replacement_field in dpg.get_item_children("__visual_editor_fields", 1): value = dpg.get_value(replacement_field) placeholder = dpg.get_item_user_data(replacement_field).placeholder query_text = query_text.replace(f"{{{{ {placeholder} }}}}", value) query_result = query.perform_query(endpoint, prefix_text + "\n" + query_text, use_post_method) except urllib.error.HTTPError as e: logging.exception(e) if e.code in [400]: set_status_text(f"Invalid query.", StatusMessageType.ERROR) return except URLError as e: logging.error(f"Connection to '{endpoint}' failed.") logging.exception(e) set_status_text(f"Connection to '{endpoint}' failed.", StatusMessageType.ERROR) return except ssl.SSLCertVerificationError: import certifi logging.error("Certificate error.") logging.error("Please make sure you have intermediate certificates installed.") logging.error(certifi.where()) set_status_text(f"Please make sure you have intermediate certificates installed. \n{certifi.where()}", StatusMessageType.ERROR) return if use_post_method: logging.debug(f"{query_result}") if query_result: set_status_text(f"{query_result}") return result_items = query_result["results"]["bindings"] if not result_items: dpg.delete_item(RESULT_TABLE_TAG, children_only=True) logging.debug(f"No results returned.") set_status_text("No results.") return dpg.delete_item(RESULT_TABLE_TAG, children_only=True) set_status_text("Query successful.") columns = result_items[0].keys() for column in columns: dpg.add_table_column(label=column, width_fixed=True, parent=RESULT_TABLE_TAG) # dpg.add_table_column(label="CCC", width_stretch=True, init_width_or_weight=0.0) # TODO This is as brittle as my ego, needs a fix asap # TODO Fix filter # with dpg.filter_set(tag="__result_filter", parent=QUERY_RESULT_TAB_TAG): for result in result_items: with dpg.table_row(parent=RESULT_TABLE_TAG): for key, value in result.items(): shortened = shorten(f"{value.get('value')}") dpg.add_text(shortened, filter_key=shortened) with dpg.drag_payload(parent=dpg.last_item(), drag_data=shortened, payload_type="string"): dpg.add_text(shortened, filter_key=shortened) def _select_directory(sender: int, mode: str, user_data: Optional[dict]): with dpg.file_dialog(directory_selector=True, show=True, callback=_process_directory): dpg.add_file_extension(".*") def _process_directory(sender: int, app_data: str, user_data: Optional[dict]): directory = app_data["file_path_name"] # files = listdir(directory) # dpg.configure_item("files_listbox", items=files) # dpg.set_value("file_text", directory) def _select_file(sender: int, app_data: str, user_data: Optional[dict]): selected_file = app_data cwd = dpg.get_value("file_text") selected_file = cwd + "/" + selected_file dpg.set_value("file_info_n", "file :" + selected_file) def create_query_group(configuration: Configuration, palette: List[QueryTemplate]): with dpg.group(horizontal=True, label="Main", tag=MAIN_QUERY_GROUP_TAG, show=True): create_query_palette(palette) with dpg.group(horizontal=False): create_query_options(configuration.endpoint_sparql) create_query_editor_visual(configuration.example_prefix, configuration.example_query, show=True) create_query_editor_textual(configuration.example_prefix, configuration.example_query, show=False) create_status_console() create_query_results()