# Copyright (c) 2024 Cisco and/or its affiliates. # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at: # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Plotly Dash HTML layout override. """ import logging import pandas as pd import dash_bootstrap_components as dbc from flask import Flask from dash import dcc from dash import html from dash import callback_context, no_update from dash import Input, Output, State from dash.exceptions import PreventUpdate from yaml import load, FullLoader, YAMLError from ..utils.constants import Constants as C from ..utils.control_panel import ControlPanel from ..utils.utils import show_tooltip, gen_new_url, get_ttypes, get_cadences, \ get_test_beds, get_job, generate_options, set_job_params, navbar_trending from ..utils.url_processing import url_decode from .graphs import graph_statistics, select_data class Layout: """The layout of the dash app and the callbacks. """ def __init__( self, app: Flask, data_stats: pd.DataFrame, data_trending: pd.DataFrame, html_layout_file: str, graph_layout_file: str, tooltip_file: str ) -> None: """Initialization: - save the input parameters, - read and pre-process the data, - prepare data for the control panel, - read HTML layout file, - read tooltips from the tooltip file. :param app: Flask application running the dash application. :param data_stats: Pandas dataframe with staistical data. :param data_trending: Pandas dataframe with trending data. :param html_layout_file: Path and name of the file specifying the HTML layout of the dash application. :param graph_layout_file: Path and name of the file with layout of plot.ly graphs. :param tooltip_file: Path and name of the yaml file specifying the tooltips. :type app: Flask :type data_stats: pandas.DataFrame :type data_trending: pandas.DataFrame :type html_layout_file: str :type graph_layout_file: str :type tooltip_file: str """ # Inputs self._app = app self._html_layout_file = html_layout_file self._graph_layout_file = graph_layout_file self._tooltip_file = tooltip_file # Pre-process the data: data_stats = data_stats[~data_stats.job.str.contains("-verify-")] data_stats = data_stats[~data_stats.job.str.contains("-coverage-")] data_stats = data_stats[~data_stats.job.str.contains("-iterative-")] data_stats = data_stats[["job", "build", "start_time", "duration"]] jobs = sorted(list(data_stats["job"].unique())) d_job_info = { "job": list(), "dut": list(), "ttype": list(), "cadence": list(), "tbed": list() } for job in jobs: idx = -3 if "-x-" in job else -2 lst_job = job.split("-") d_job_info["job"].append(job) d_job_info["dut"].append(lst_job[1]) d_job_info["ttype"].append(lst_job[3]) d_job_info["cadence"].append(lst_job[4]) d_job_info["tbed"].append("-".join(lst_job[idx:])) self._job_info = pd.DataFrame.from_dict(d_job_info) self._default = set_job_params(self._job_info, d_job_info["job"][0]) tst_info = { "job": list(), "build": list(), "dut_type": list(), "dut_version": list(), "hosts": list(), "passed": list(), "failed": list(), "lst_failed": list() } for job in jobs: df_job = data_trending.loc[(data_trending["job"] == job)] builds = df_job["build"].unique() for build in builds: df_build = df_job.loc[(df_job["build"] == build)] tst_info["job"].append(job) tst_info["build"].append(build) tst_info["dut_type"].append(df_build["dut_type"].iloc[-1]) tst_info["dut_version"].append(df_build["dut_version"].iloc[-1]) tst_info["hosts"].append(df_build["hosts"].iloc[-1]) try: passed = df_build.value_counts(subset="passed")[True] except KeyError: passed = 0 try: failed = df_build.value_counts(subset="passed")[False] failed_tests = df_build.loc[(df_build["passed"] == False)]\ ["test_id"].to_list() l_failed = list() for tst in failed_tests: lst_tst = tst.split(".") suite = lst_tst[-2].replace("2n1l-", "").\ replace("1n1l-", "").replace("2n-", "") l_failed.append(f"{suite.split('-')[0]}-{lst_tst[-1]}") except KeyError: failed = 0 l_failed = list() tst_info["passed"].append(passed) tst_info["failed"].append(failed) tst_info["lst_failed"].append(sorted(l_failed)) self._data = data_stats.merge(pd.DataFrame.from_dict(tst_info)) # Read from files: self._html_layout = str() self._graph_layout = None self._tooltips = dict() try: with open(self._html_layout_file, "r") as file_read: self._html_layout = file_read.read() except IOError as err: raise RuntimeError( f"Not possible to open the file {self._html_layout_file}\n{err}" ) try: with open(self._graph_layout_file, "r") as file_read: self._graph_layout = load(file_read, Loader=FullLoader) except IOError as err: raise RuntimeError( f"Not possible to open the file {self._graph_layout_file}\n" f"{err}" ) except YAMLError as err: raise RuntimeError( f"An error occurred while parsing the specification file " f"{self._graph_layout_file}\n{err}" ) try: with open(self._tooltip_file, "r") as file_read: self._tooltips = load(file_read, Loader=FullLoader) except IOError as err: logging.warning( f"Not possible to open the file {self._tooltip_file}\n{err}" ) except YAMLError as err: logging.warning( f"An error occurred while parsing the specification file " f"{self._tooltip_file}\n{err}" ) # Control panel partameters and their default values. self._cp_default = { "ri-ttypes-options": self._default["ttypes"], "ri-cadences-options": self._default["cadences"], "dd-tbeds-options": self._default["tbeds"], "ri-duts-value": self._default["dut"], "ri-ttypes-value": self._default["ttype"], "ri-cadences-value": self._default["cadence"], "dd-tbeds-value": self._default["tbed"], "al-job-children": html.A( self._default["job"], href=f"{C.URL_CICD}{self._default['job']}", target="_blank" ) } # Callbacks: if self._app is not None and hasattr(self, "callbacks"): self.callbacks(self._app) @property def html_layout(self) -> dict: return self._html_layout def add_content(self): """Top level method which generated the web page. It generates: - Store for user input data, - Navigation bar, - Main area with control panel and ploting area. If no HTML layout is provided, an error message is displayed instead. :returns: The HTML div with the whole page. :rtype: html.Div """ if self.html_layout: return html.Div( id="div-main", className="small", children=[ dcc.Store(id="control-panel"), dcc.Location(id="url", refresh=False), dbc.Row( id="row-navbar", class_name="g-0", children=[navbar_trending((False, False, True, False))] ), dbc.Spinner( dbc.Offcanvas( class_name="w-50", id="offcanvas-metadata", title="Detailed Information", placement="end", is_open=False, children=[ dbc.Row(id="row-metadata") ] ) ), dbc.Row( id="row-main", class_name="g-0", children=[ self._add_ctrl_col(), self._add_plotting_col() ] ), dbc.Offcanvas( class_name="w-75", id="offcanvas-documentation", title="Documentation", placement="end", is_open=False, children=html.Iframe( src=C.URL_DOC_TRENDING, width="100%", height="100%" ) ) ] ) else: return html.Div( id="div-main-error", children=[ dbc.Alert( [ "An Error Occured", ], color="danger" ) ] ) def _add_ctrl_col(self) -> dbc.Col: """Add column with controls. It is placed on the left side. :returns: Column with the control panel. :rtype: dbc.Col """ return dbc.Col([ html.Div( children=self._add_ctrl_panel(), className="sticky-top" ) ]) def _add_plotting_col(self) -> dbc.Col: """Add column with plots and tables. It is placed on the right side. :returns: Column with tables. :rtype: dbc.Col """ return dbc.Col( id="col-plotting-area", children=[ dbc.Spinner( children=[ dbc.Row( id="plotting-area", class_name="g-0 p-0", children=[ C.PLACEHOLDER ] ) ] ) ], width=9 ) def _add_ctrl_panel(self) -> dbc.Row: """Add control panel. :returns: Control panel. :rtype: dbc.Row """ return [ dbc.Row( class_name="g-0 p-1", children=[ dbc.InputGroup( [ dbc.InputGroupText(show_tooltip( self._tooltips, "help-dut", "DUT" )), dbc.RadioItems( id="ri-duts", inline=True, value=self._default["dut"], options=self._default["duts"], class_name="form-control" ) ], size="sm" ) ] ), dbc.Row( class_name="g-0 p-1", children=[ dbc.InputGroup( [ dbc.InputGroupText(show_tooltip( self._tooltips, "help-ttype", "Test Type" )), dbc.RadioItems( id="ri-ttypes", inline=True, value=self._default["ttype"], options=self._default["ttypes"], class_name="form-control" ) ], size="sm" ) ] ), dbc.Row( class_name="g-0 p-1", children=[ dbc.InputGroup( [ dbc.InputGroupText(show_tooltip( self._tooltips, "help-cadence", "Cadence" )), dbc.RadioItems( id="ri-cadences", inline=True, value=self._default["cadence"], options=self._default["cadences"], class_name="form-control" ) ], size="sm" ) ] ), dbc.Row( class_name="g-0 p-1", children=[ dbc.InputGroup( [ dbc.InputGroupText(show_tooltip( self._tooltips, "help-tbed", "Test Bed" )), dbc.Select( id="dd-tbeds", placeholder="Select a test bed...", value=self._default["tbed"], options=self._default["tbeds"] ) ], size="sm" ) ] ), dbc.Row( class_name="g-0 p-1", children=[ dbc.Alert( id="al-job", color="info", children=self._default["job"] ) ] ) ] def _get_plotting_area( self, job: str, url: str ) -> list: """Generate the plotting area with all its content. :param job: The job which data will be displayed. :param url: URL to be displayed in the modal window. :type job: str :type url: str :returns: List of rows with elements to be displayed in the plotting area. :rtype: list """ figs = graph_statistics(self._data, job, self._graph_layout) if not figs[0]: return C.PLACEHOLDER return [ dbc.Row( id="row-graph-passed", class_name="g-0 p-1", children=[ dcc.Graph( id="graph-passed", figure=figs[0] ) ] ), dbc.Row( id="row-graph-duration", class_name="g-0 p-1", children=[ dcc.Graph( id="graph-duration", figure=figs[1] ) ] ), dbc.Row( [ dbc.Col([html.Div( [ dbc.Button( id="plot-btn-url", children="Show URL", class_name="me-1", color="info", style={ "text-transform": "none", "padding": "0rem 1rem" } ), dbc.Modal( [ dbc.ModalHeader(dbc.ModalTitle("URL")), dbc.ModalBody(url) ], id="plot-mod-url", size="xl", is_open=False, scrollable=True ), dbc.Button( id="plot-btn-download", children="Download Data", class_name="me-1", color="info", style={ "text-transform": "none", "padding": "0rem 1rem" } ), dcc.Download(id="download-stats-data") ], className=\ "d-grid gap-0 d-md-flex justify-content-md-end" )]) ], class_name="g-0 p-0" ) ] def callbacks(self, app): """Callbacks for the whole application. :param app: The application. :type app: Flask """ @app.callback( Output("control-panel", "data"), # Store Output("plotting-area", "children"), Output("ri-ttypes", "options"), Output("ri-cadences", "options"), Output("dd-tbeds", "options"), Output("ri-duts", "value"), Output("ri-ttypes", "value"), Output("ri-cadences", "value"), Output("dd-tbeds", "value"), Output("al-job", "children"), State("control-panel", "data"), # Store Input("ri-duts", "value"), Input("ri-ttypes", "value"), Input("ri-cadences", "value"), Input("dd-tbeds", "value"), Input("url", "href") ) def _update_ctrl_panel(cp_data: dict, dut: str, ttype: str, cadence: str, tbed: str, href: str) -> tuple: """Update the application when the event is detected. :param cp_data: Current status of the control panel stored in browser. :param dut: Input - DUT name. :param ttype: Input - Test type. :param cadence: Input - The cadence of the job. :param tbed: Input - The test bed. :param href: Input - The URL provided by the browser. :type cp_data: dict :type dut: str :type ttype: str :type cadence: str :type tbed: str :type href: str :returns: New values for web page elements. :rtype: tuple """ ctrl_panel = ControlPanel(self._cp_default, cp_data) # Parse the url: parsed_url = url_decode(href) if parsed_url: url_params = parsed_url["params"] else: url_params = None trigger_id = callback_context.triggered[0]["prop_id"].split(".")[0] if trigger_id == "ri-duts": ttype_opts = generate_options(get_ttypes(self._job_info, dut)) ttype_val = ttype_opts[0]["value"] cad_opts = generate_options(get_cadences( self._job_info, dut, ttype_val)) cad_val = cad_opts[0]["value"] tbed_opts = generate_options(get_test_beds( self._job_info, dut, ttype_val, cad_val)) tbed_val = tbed_opts[0]["value"] ctrl_panel.set({ "ri-duts-value": dut, "ri-ttypes-options": ttype_opts, "ri-ttypes-value": ttype_val, "ri-cadences-options": cad_opts, "ri-cadences-value": cad_val, "dd-tbeds-options": tbed_opts, "dd-tbeds-value": tbed_val }) elif trigger_id == "ri-ttypes": cad_opts = generate_options(get_cadences( self._job_info, ctrl_panel.get("ri-duts-value"), ttype)) cad_val = cad_opts[0]["value"] tbed_opts = generate_options(get_test_beds( self._job_info, ctrl_panel.get("ri-duts-value"), ttype, cad_val)) tbed_val = tbed_opts[0]["value"] ctrl_panel.set({ "ri-ttypes-value": ttype, "ri-cadences-options": cad_opts, "ri-cadences-value": cad_val, "dd-tbeds-options": tbed_opts, "dd-tbeds-value": tbed_val }) elif trigger_id == "ri-cadences": tbed_opts = generate_options(get_test_beds( self._job_info, ctrl_panel.get("ri-duts-value"), ctrl_panel.get("ri-ttypes-value"), cadence)) tbed_val = tbed_opts[0]["value"] ctrl_panel.set({ "ri-cadences-value": cadence, "dd-tbeds-options": tbed_opts, "dd-tbeds-value": tbed_val }) elif trigger_id == "dd-tbeds": ctrl_panel.set({ "dd-tbeds-value": tbed }) elif trigger_id == "url": if url_params: new_job = url_params.get("job", list())[0] if new_job: job_params = set_job_params(self._job_info, new_job) ctrl_panel = ControlPanel( { "ri-ttypes-options": job_params["ttypes"], "ri-cadences-options": job_params["cadences"], "dd-tbeds-options": job_params["tbeds"], "ri-duts-value": job_params["dut"], "ri-ttypes-value": job_params["ttype"], "ri-cadences-value": job_params["cadence"], "dd-tbeds-value": job_params["tbed"], "al-job-children": html.A( self._default["job"], href=( f"{C.URL_CICD}" f"{self._default['job']}" ), target="_blank" ) }, None ) else: ctrl_panel = ControlPanel(self._cp_default, cp_data) job = get_job( self._job_info, ctrl_panel.get("ri-duts-value"), ctrl_panel.get("ri-ttypes-value"), ctrl_panel.get("ri-cadences-value"), ctrl_panel.get("dd-tbeds-value") ) ctrl_panel.set( { "al-job-children": html.A( job, href=f"{C.URL_CICD}{job}", target="_blank" ) } ) plotting_area = self._get_plotting_area( job, gen_new_url(parsed_url, {"job": job}) ) ret_val = [ ctrl_panel.panel, plotting_area ] ret_val.extend(ctrl_panel.values) return ret_val @app.callback( Output("plot-mod-url", "is_open"), [Input("plot-btn-url", "n_clicks")], [State("plot-mod-url", "is_open")], ) def toggle_plot_mod_url(n, is_open): """Toggle the modal window with url. """ if n: return not is_open return is_open @app.callback( Output("download-stats-data", "data"), State("control-panel", "data"), # Store Input("plot-btn-download", "n_clicks"), prevent_initial_call=True ) def _download_data(cp_data: dict, n_clicks: int): """Download the data :param cp_data: Current status of the control panel stored in browser. :param n_clicks: Number of clicks on the button "Download". :type cp_data: dict :type n_clicks: int :returns: dict of data frame content (base64 encoded) and meta data used by the Download component. :rtype: dict """ if not n_clicks: raise PreventUpdate ctrl_panel = ControlPanel(self._cp_default, cp_data) job = get_job( self._job_info, ctrl_panel.get("ri-duts-value"), ctrl_panel.get("ri-ttypes-value"), ctrl_panel.get("ri-cadences-value"), ctrl_panel.get("dd-tbeds-value") ) data = select_data(self._data, job) data = data.drop(columns=["job", ]) return dcc.send_data_frame( data.T.to_csv, f"{job}-{C.STATS_DOWNLOAD_FILE_NAME}") @app.callback( Output("row-metadata", "children"), Output("offcanvas-metadata", "is_open"), Input("graph-passed", "clickData"), Input("graph-duration", "clickData"), prevent_initial_call=True ) def _show_metadata_from_graphs( passed_data: dict, duration_data: dict) -> tuple: """Generates the data for the offcanvas displayed when a particular point in a graph is clicked on. :param passed_data: The data from the clicked point in the graph displaying the pass/fail data. :param duration_data: The data from the clicked point in the graph displaying the duration data. :type passed_data: dict :type duration data: dict :returns: The data to be displayed on the offcanvas (job statistics and the list of failed tests) and the information to show the offcanvas. :rtype: tuple(list, bool) """ if not (passed_data or duration_data): raise PreventUpdate metadata = no_update open_canvas = False title = "Job Statistics" trigger_id = callback_context.triggered[0]["prop_id"].split(".")[0] if trigger_id == "graph-passed": graph_data = passed_data["points"][0].get("hovertext", "") elif trigger_id == "graph-duration": graph_data = duration_data["points"][0].get("text", "") if graph_data: lst_graph_data = graph_data.split("
") # Prepare list of failed tests: job = str() build = str() for itm in lst_graph_data: if "csit-ref:" in itm: job, build = itm.split(" ")[-1].split("/") break if job and build: fail_tests = self._data.loc[ (self._data["job"] == job) & (self._data["build"] == build) ]["lst_failed"].values[0] if not fail_tests: fail_tests = None else: fail_tests = None # Create the content of the offcanvas: list_group_items = list() for itm in lst_graph_data: lst_itm = itm.split(": ") if lst_itm[0] == "csit-ref": list_group_item = dbc.ListGroupItem([ dbc.Badge(lst_itm[0]), html.A( lst_itm[1], href=f"{C.URL_LOGS}{lst_itm[1]}", target="_blank" ) ]) else: list_group_item = dbc.ListGroupItem([ dbc.Badge(lst_itm[0]), lst_itm[1] ]) list_group_items.append(list_group_item) metadata = [ dbc.Card( class_name="gy-2 p-0", children=[ dbc.CardHeader([ dcc.Clipboard( target_id="metadata", title="Copy", style={"display": "inline-block"} ), title ]), dbc.CardBody( dbc.ListGroup(list_group_items, flush=True), id="metadata", class_name="p-0" ) ] ) ] if fail_tests is not None: metadata.append( dbc.Card( class_name="gy-2 p-0", children=[ dbc.CardHeader( f"List of Failed Tests ({len(fail_tests)})" ), dbc.CardBody( id="failed-tests", class_name="p-0", children=[dbc.ListGroup( children=[ dbc.ListGroupItem(x) \ for x in fail_tests ], flush=True), ] ) ] ) ) open_canvas = True return metadata, open_canvas @app.callback( Output("offcanvas-documentation", "is_open"), Input("btn-documentation", "n_clicks"), State("offcanvas-documentation", "is_open") ) def toggle_offcanvas_documentation(n_clicks, is_open): if n_clicks: return not is_open return is_open