Spaces:
Running
Running
| from utils import * | |
| current_path = pathlib.Path(__file__).parent.absolute().as_posix() | |
| configuration = get_obsei_config(current_path, "config.yaml") | |
| logo_url = "https://raw.githubusercontent.com/obsei/obsei-resources/master/logos/obsei_200x200.png" | |
| st.set_page_config(page_title="Obsei Demo", layout="wide", page_icon=logo_url) | |
| st.title("Obsei Demo").markdown( | |
| get_icon_name("Obsei Demo", logo_url, 60, 35), unsafe_allow_html=True | |
| ) | |
| st.error( | |
| """ | |
| **Note:** Demo run will require some secure information based on source or sink selected, | |
| if you don't trust this environment please close the app.\n | |
| Do not share and commit generated file as it may contains secure information. | |
| """ | |
| ) | |
| ( | |
| pipeline_col, | |
| spinner_col, | |
| execute_col, | |
| download_python_col, | |
| download_yaml_col, | |
| ) = st.columns([2, 2, 1, 1, 1]) | |
| source_col, analyzer_col, sink_col = st.columns([1, 1, 1]) | |
| source_list = [k for k in configuration["source"].keys()] | |
| selected_source = source_col.selectbox("Select Observer", source_list) | |
| analyzer_list = [k for k in configuration["analyzer"].keys()] | |
| selected_analyzer = analyzer_col.selectbox("Select Analyzer", analyzer_list) | |
| sink_list = [k for k in configuration["sink"].keys()] | |
| selected_sink = sink_col.selectbox("Select Informer", sink_list) | |
| src_icon = get_icon_name(None, configuration["source"][selected_source]["_icon_"]) | |
| analyzer_icon = get_icon_name( | |
| None, configuration["analyzer"][selected_analyzer]["_icon_"] | |
| ) | |
| sink_icon = get_icon_name(None, configuration["sink"][selected_sink]["_icon_"]) | |
| pipeline_col.header("Pipeline").markdown( | |
| f"**Pipeline:** {src_icon} ➡➡ {analyzer_icon} ➡➡ {sink_icon}", | |
| unsafe_allow_html=True, | |
| ) | |
| src_config = configuration["source"][selected_source]["config"] | |
| render_config( | |
| src_config, source_col, configuration["source"][selected_source]["_help_"] | |
| ) | |
| analyzer_type_config = configuration["analyzer"][selected_analyzer] | |
| analyzer_type_list = [] | |
| for k in analyzer_type_config.keys(): | |
| if k != "_icon_": | |
| analyzer_type_list.append(k) | |
| selected_analyzer_type = analyzer_col.selectbox(f"analyzer type", analyzer_type_list) | |
| analyzer_config = None | |
| if "config" in analyzer_type_config[selected_analyzer_type]: | |
| analyzer_config = analyzer_type_config[selected_analyzer_type]["config"] | |
| render_config( | |
| analyzer_config, | |
| analyzer_col, | |
| analyzer_type_config[selected_analyzer_type]["_help_"], | |
| ) | |
| if len(analyzer_type_config[selected_analyzer_type]["analyzer"]) > 1: | |
| render_config( | |
| analyzer_type_config[selected_analyzer_type]["analyzer"], analyzer_col | |
| ) | |
| sink_config = configuration["sink"][selected_sink]["config"] | |
| render_config(sink_config, sink_col, configuration["sink"][selected_sink]["_help_"]) | |
| generate_config = { | |
| "source": configuration["source"][selected_source]["source"], | |
| "source_config": src_config, | |
| "analyzer_config": analyzer_config, | |
| "analyzer": analyzer_type_config[selected_analyzer_type]["analyzer"], | |
| "sink": configuration["sink"][selected_sink]["sink"], | |
| "sink_config": sink_config, | |
| } | |
| python_code = generate_python(generate_config) | |
| yaml_code = generate_yaml(generate_config) | |
| execute_button = execute_col.button("🚀 Execute") | |
| if execute_button: | |
| execute_workflow(generate_config, spinner_col) | |
| with download_python_col: | |
| download_button(python_code, "generated-code.py", "🐍 Download (.py)") | |
| with download_yaml_col: | |
| download_button(yaml_code, "generated-config.yaml", "📖 Download (.yaml)") | |