import time import webbrowser from pathlib import Path from threading import Thread from typing import Any, Final, cast import dash_cytoscape as cyto import plotly.express as px from dash import ( Dash, Input, Output, State, callback, dash_table, dcc, html, ) from pandas import DataFrame import lang_main.io from lang_main.analysis import graphs, tokens from lang_main.constants import SAVE_PATH_FOLDER, SPCY_MODEL from lang_main.types import EntryPoints, ObjectID, TimelineCandidates # ** data # p_df = Path(r'../results/test_20240619/TIMELINE.pkl').resolve() p_df = lang_main.io.get_entry_point(SAVE_PATH_FOLDER, EntryPoints.TIMELINE) (data,) = cast(tuple[DataFrame], lang_main.io.load_pickle(p_df)) # p_tl = Path(r'../results/test_20240619/TIMELINE_POSTPROCESSING.pkl').resolve() p_tl = lang_main.io.get_entry_point(SAVE_PATH_FOLDER, EntryPoints.TIMELINE_POST) cands, texts = cast( tuple[TimelineCandidates, dict[ObjectID, str]], lang_main.io.load_pickle(p_tl) ) TABLE_FEATS: Final[list[str]] = [ 'ErstellungsDatum', 'ErledigungsDatum', 'VorgangsTypName', 'VorgangsBeschreibung', ] TABLE_FEATS_DATES: Final[list[str]] = [ 'ErstellungsDatum', 'ErledigungsDatum', ] # ** figure config MARKERS: Final[dict[str, Any]] = { 'size': 12, 'color': 'yellow', 'line': { 'width': 2, 'color': 'red', }, } HOVER_DATA: Final[dict[str, Any]] = { 'ErstellungsDatum': '|%d.%m.%Y', 'VorgangsBeschreibung': True, } # ** graph # target = '../results/test_20240529/Pipe-Token_Analysis_Step-1_build_token_graph.pkl' # p = Path(target).resolve() p_tk_graph = lang_main.io.get_entry_point(SAVE_PATH_FOLDER, EntryPoints.TK_GRAPH_POST) ret = lang_main.io.load_pickle(p_tk_graph) tk_graph = cast(graphs.TokenGraph, ret[0]) tk_graph_filtered = graphs.filter_graph_by_edge_weight(tk_graph, 150, None) tk_graph_filtered = graphs.filter_graph_by_node_degree(tk_graph_filtered, 1, None) # tk_graph_filtered = tk_graph.filter_by_edge_weight(150, None) # tk_graph_filtered = tk_graph_filtered.filter_by_node_degree(1, None) cyto_data_base, weight_data = graphs.convert_graph_to_cytoscape(tk_graph_filtered) MIN_WEIGHT = weight_data['min'] MAX_WEIGHT = weight_data['max'] cyto.load_extra_layouts() cose_layout = { 'name': 'cose', 'nodeOverlap': 500, 'refresh': 20, 'fit': True, 'padding': 20, 'randomize': False, 'componentSpacing': 1.2, 'nodeRepulsion': 1000, 'edgeElasticity': 1000, 'idealEdgeLength': 100, 'nestingFactor': 1.2, 'gravity': 50, 'numIter': 3000, 'initialTemp': 2000, 'coolingFactor': 0.7, 'minTemp': 1.0, 'nodeDimensionsIncludeLabels': True, } my_stylesheet = [ # Group selectors { 'selector': 'node', 'style': { 'shape': 'circle', 'content': 'data(label)', 'background-color': '#B10DC9', 'border-width': 2, 'border-color': 'black', 'border-opacity': 1, 'opacity': 1, 'color': 'black', 'text-opacity': 1, 'font-size': 12, 'z-index': 9999, }, }, { 'selector': 'edge', 'style': { #'width': f'mapData(weight, {MIN_WEIGHT}, {MAX_WEIGHT}, 1, 10)', # 'width': """function(ele) { # return ele.data('weight'); # """, 'curve-style': 'bezier', 'line-color': 'grey', 'line-style': 'solid', 'line-opacity': 1, }, }, # Class selectors # {'selector': '.red', 'style': {'background-color': 'red', 'line-color': 'red'}}, # {'selector': '.triangle', 'style': {'shape': 'triangle'}}, ] graph_layout = html.Div( [ html.Button('Trigger JS Weight', id='test_js_weight'), html.Button('Trigger Candidate Graph', id='graph-build-btn'), dcc.Store(id='graph-store', storage_type='memory'), dcc.Store(id='graph-store-cyto-curr_cands', storage_type='memory'), html.Div(id='output'), html.Div( [ html.H2('Token Graph', style={'margin': 0}), html.Button( 'Reset Default', id='bt-reset', style={ 'marginLeft': 'auto', 'width': '300px', }, ), ], style={ 'display': 'flex', 'marginBottom': '1em', }, ), html.H3('Layout'), dcc.Dropdown( id='layout_choice', options=[ 'cose', 'cola', 'euler', 'random', ], value='cose', clearable=False, ), html.Div( [ html.H3('Graph Filter'), dcc.Input( id='graph-weight_min', type='number', min=MIN_WEIGHT, max=MAX_WEIGHT, step=1, placeholder=f'Minimum edge weight: {MIN_WEIGHT} - {MAX_WEIGHT}', debounce=True, style={'width': '40%'}, ), dcc.Input( id='graph-weight_max', type='number', min=MIN_WEIGHT, max=MAX_WEIGHT, step=1, placeholder=f'Maximum edge weight: {MIN_WEIGHT} - {MAX_WEIGHT}', debounce=True, style={'width': '40%'}, ), html.H3('Graph'), html.Button('Re-Layout', id='graph-trigger_relayout'), html.Div( [ cyto.Cytoscape( id='cytoscape-graph', style={'width': '100%', 'height': '600px'}, layout=cose_layout, stylesheet=my_stylesheet, elements=cyto_data_base, zoom=1, ), ], style={ 'border': '3px solid black', 'borderRadius': '25px', 'marginTop': '1em', 'marginBottom': '2em', 'padding': '7px', }, ), ], style={'marginTop': '1em'}, ), ], ) # ** app external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = Dash(__name__, external_stylesheets=external_stylesheets) app.layout = html.Div( [ html.H1(children='Demo Zeitreihenanalyse', style={'textAlign': 'center'}), html.Div( children=[ html.H2('Wählen Sie ein Objekt aus (ObjektID):'), dcc.Dropdown( list(cands.keys()), id='selector-obj_id', placeholder='ObjektID auswählen...', ), ] ), html.Div( children=[ html.H3(id='object-text'), dcc.Dropdown(id='selector-candidates'), dcc.Graph(id='graph-candidates'), ] ), html.Div( [dash_table.DataTable(id='table-candidates')], style={'marginBottom': '2em'} ), graph_layout, ], style={'margin': '2em'}, ) @callback( Output('object-text', 'children'), Input('selector-obj_id', 'value'), prevent_initial_call=True, ) def update_obj_text(obj_id): obj_id = int(obj_id) obj_text = texts[obj_id] headline = f'HObjektText: {obj_text}' return headline @callback( [Output('selector-candidates', 'options'), Output('selector-candidates', 'value')], Input('selector-obj_id', 'value'), prevent_initial_call=True, ) def update_choice_candidates(obj_id): obj_id = int(obj_id) cands_obj_id = cands[obj_id] choices = list(range(1, len(cands_obj_id) + 1)) return choices, choices[0] # TODO check possible storage of pre-filtered result # TODO change input of ``update_table_candidates`` and ``display_candidates_as_graph`` # TODO to storage component @callback( Output('graph-candidates', 'figure'), Input('selector-candidates', 'value'), State('selector-obj_id', 'value'), prevent_initial_call=True, ) def update_timeline(index, obj_id): obj_id = int(obj_id) # title obj_text = texts[obj_id] title = f'HObjektText: {obj_text}' # cands # cands_per_obj_id = cands[obj_id] # cands_similar = cands_per_obj_id[int(index) - 1] # data # df = data.loc[list(cands_similar)].sort_index() # type: ignore df = pre_filter_data(data, idx=index, obj_id=obj_id) # figure fig = px.line( data_frame=df, x='ErstellungsDatum', y='ObjektID', title=title, hover_data=HOVER_DATA, ) fig.update_traces(mode='markers+lines', marker=MARKERS, marker_symbol='diamond') fig.update_xaxes( tickformat='%B\n%Y', rangeslider_visible=True, ) fig.update_yaxes(type='category') fig.update_layout(hovermode='x unified') return fig @callback( [Output('table-candidates', 'data'), Output('table-candidates', 'columns')], Input('selector-candidates', 'value'), State('selector-obj_id', 'value'), prevent_initial_call=True, ) def update_table_candidates(index, obj_id): df = pre_filter_data(data, idx=index, obj_id=obj_id) df = df.filter(items=TABLE_FEATS, axis=1).sort_values( by='ErstellungsDatum', ascending=True ) cols = [{'name': i, 'id': i} for i in df.columns] # convert dates to strings for col in TABLE_FEATS_DATES: df[col] = df[col].dt.strftime(r'%Y-%m-%d') table_data = df.to_dict('records') return table_data, cols def pre_filter_data( data: DataFrame, idx: int, obj_id: ObjectID, ) -> DataFrame: idx = int(idx) obj_id = int(obj_id) data = data.copy() # cands cands_obj_id = cands[obj_id] cands_choice = cands_obj_id[int(idx) - 1] # data data = data.loc[list(cands_choice)].sort_index() # type: ignore return data # ** graph callbacks # TODO store pre-calculated graph @app.callback( Output('cytoscape-graph', 'elements', allow_duplicate=True), Output('graph-weight_min', 'min', allow_duplicate=True), Output('graph-weight_min', 'max', allow_duplicate=True), Output('graph-weight_min', 'placeholder', allow_duplicate=True), Output('graph-weight_max', 'min', allow_duplicate=True), Output('graph-weight_max', 'max', allow_duplicate=True), Output('graph-weight_max', 'placeholder', allow_duplicate=True), Output('graph-store', 'data'), Output('graph-store-cyto-curr_cands', 'data'), # Input('graph-build-btn', 'n_clicks'), Input('selector-candidates', 'value'), State('selector-obj_id', 'value'), prevent_initial_call=True, ) def display_candidates_as_graph(index, obj_id): t1 = time.perf_counter() df = pre_filter_data(data, idx=index, obj_id=obj_id) t2 = time.perf_counter() print(f'Time for filtering: {t2 - t1} s') t1 = time.perf_counter() tk_graph_cands, _ = tokens.build_token_graph( data=df, model=SPCY_MODEL, target_feature='VorgangsBeschreibung', build_map=False, logging_graph=False, ) t2 = time.perf_counter() print(f'Time for graph building: {t2 - t1} s') t1 = time.perf_counter() cyto_data, weight_info = graphs.convert_graph_to_cytoscape(tk_graph_cands) weight_min = weight_info['min'] weight_max = weight_info['max'] placeholder_min = f'Minimum edge weight: {weight_min} - {weight_max}' placeholder_max = f'Maximum edge weight: {weight_min} - {weight_max}' t2 = time.perf_counter() print(f'Time for graph metadata and conversion: {t2 - t1} s') t1 = time.perf_counter() graph_to_store = lang_main.io.encode_to_base64_str(tk_graph_cands) t2 = time.perf_counter() print(f'Time for encoding: {t2 - t1} s') return ( cyto_data, weight_min, weight_max, placeholder_min, weight_min, weight_max, placeholder_max, graph_to_store, cyto_data, ) @app.callback( Output('cytoscape-graph', 'layout', allow_duplicate=True), Input('layout_choice', 'value'), prevent_initial_call=True, ) def update_layout_internal(layout_choice): # return {'name': layout_choice} return cose_layout # return cose_bilkent_layout # return cola_layout @app.callback( Output('cytoscape-graph', 'zoom'), Output('cytoscape-graph', 'elements', allow_duplicate=True), Output('graph-weight_min', 'value'), Output('graph-weight_max', 'value'), Input('bt-reset', 'n_clicks'), State('graph-store-cyto-curr_cands', 'data'), prevent_initial_call=True, ) def reset_layout(_, current_cands_cyto_elements): return (1, current_cands_cyto_elements, None, None) # update edge weight @app.callback( Output('cytoscape-graph', 'elements', allow_duplicate=True), Input('graph-weight_min', 'value'), Input('graph-weight_max', 'value'), State('graph-store', 'data'), State('graph-store-cyto-curr_cands', 'data'), State('graph-weight_min', 'min'), State('graph-weight_min', 'max'), prevent_initial_call=True, ) def update_edge_weight( weight_min, weight_max, current_graph, current_cands_cyto_elements, current_min, current_max, ): if not any((weight_min, weight_max)): return current_cands_cyto_elements if weight_min is None: weight_min = current_min if weight_max is None: weight_max = current_max tk_graph = cast(graphs.TokenGraph, lang_main.io.decode_from_base64_str(current_graph)) tk_graph_filtered = graphs.filter_graph_by_edge_weight(tk_graph, weight_min, weight_max) # tk_graph_filtered = tk_graph.filter_by_edge_weight(weight_min, weight_max) tk_graph_filtered = graphs.filter_graph_by_node_degree(tk_graph_filtered, 1, None) # tk_graph_filtered = tk_graph_filtered.filter_by_node_degree(1, None) cyto_data, _ = graphs.convert_graph_to_cytoscape(tk_graph_filtered) return cyto_data # ** graph: layout with edge weight app.clientside_callback( """ function(n_clicks, layout) { layout.edgeElasticity = function(edge) { return edge.data().weight * 0.05; }; layout.idealEdgeLength = function(edge) { return edge.data().weight * 0.4; }; cy.layout(layout).run(); return layout; } """, Output('cytoscape-graph', 'layout', allow_duplicate=True), Input('graph-trigger_relayout', 'n_clicks'), State('cytoscape-graph', 'layout'), prevent_initial_call=True, ) # ** graph: display edge weight (line thickness) app.clientside_callback( """ function(n_clicks, stylesheet) { function edge_weight(ele) { let threshold = 1000; let weight = ele.data('weight'); if (weight > threshold) { weight = 12; } else { weight = weight / threshold * 10; weight = Math.max(1, weight); } return weight; } stylesheet[1].style.width = edge_weight; cy.style(stylesheet).update(); return stylesheet; } """, Output('cytoscape-graph', 'stylesheet'), Input('test_js_weight', 'n_clicks'), State('cytoscape-graph', 'stylesheet'), prevent_initial_call=False, ) def _start_webbrowser(): host = '127.0.0.1' port = '8050' adress = f'http://{host}:{port}/' time.sleep(2) webbrowser.open_new(adress) def main(): webbrowser_thread = Thread(target=_start_webbrowser, daemon=True) webbrowser_thread.start() app.run(debug=True) if __name__ == '__main__': main()