-
Notifications
You must be signed in to change notification settings - Fork 155
/
ui.py
96 lines (77 loc) · 3.41 KB
/
ui.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
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.success(
"""
Please ⭐ the repo and share the feedback at https://github.com/obsei/obsei?utm_source=streamlit
"""
)
st.warning(
"""
**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.
"""
)
(
pipeline_col,
spinner_col,
execute_col,
download_python_col,
download_yaml_col,
) = st.columns([2, 2, 1, 1, 1])
col_map = dict()
col_map["source"], col_map["analyzer"], col_map["sink"] = st.columns([1, 1, 1])
selected = {}
name_map = {"source": "Observer", "analyzer": "Analyzer", "sink": "Informer"}
for node_name, col in col_map.items():
item_list = [k for k in configuration[node_name].keys()]
selected[node_name] = col.selectbox(f"Select {name_map[node_name]}", item_list)
icons = [get_icon_name(None, configuration[k][v]["_icon_"]) for k, v in selected.items()]
pipeline_col.header("Pipeline").markdown(
f"**Pipeline:** {icons[0]} ➡➡ {icons[1]} ➡➡ {icons[2]}",
unsafe_allow_html=True,
)
generate_config = {}
log_component = {}
for node_name, node_value in selected.items():
type_config = configuration[node_name][node_value]
if node_name == "analyzer":
type_list = []
for config_key in type_config.keys():
if config_key != "_icon_":
type_list.append(config_key)
selected_type = col_map[node_name].selectbox(f"{name_map[node_name]} Type", type_list)
type_config = type_config[selected_type]
config = None
if "config" in type_config:
config = type_config["config"]
if type_config["_help_"] is not None:
with col_map[node_name].expander("Config Help Info", False):
help_area = "\n".join(type_config["_help_"])
st.code(f"{help_area}")
config_expander = None
if config is not None:
config_expander = col_map[node_name].expander(f"Configure {name_map[node_name]}", False)
render_config(config, config_expander)
if node_name == "analyzer" and node_name in type_config and len(type_config[node_name]) > 1:
config_expander = config_expander or col_map[node_name].expander(f"Configure {name_map[node_name]}", False)
render_config(type_config["analyzer"], config_expander)
generate_config[node_name] = type_config[node_name]
generate_config[f"{node_name}_config"] = config
log_expander = col_map[node_name].expander(f"{name_map[node_name]} Logs", True)
log_component[node_name] = log_expander.empty()
log_component[node_name].write("Run \"🚀 Execute\" first")
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, log_component)
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)")